﻿/*************************************************************************************
 *
 * 文 件 名： Worker
 * 描    述： 从模板新增
 * 
 * 版    本： V1.0
 * 创 建 者： YSW
 * 创建时间： 2022-08-24 8:54:29
 * ======================================
 * 历史更新记录
 * 版本：V          修改时间：         修改人：
 * 修改内容：
 * ======================================
*************************************************************************************/

using CommonModel.Model;
using CommonModel.Util;
using GXRTBTC.PublicClassLibrary;
using GXRTBTC.PublicClassLibrary.Model;
using GXRTBTC.PublicClassLibrary.Redis;
using GXRTBTC.PublicClassLibrary.Util;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
using NewLife;
using NewLife.Caching;
using NewLife.Data;
using NewLife.Http;
using NewLife.IP;
using NewLife.Messaging;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RedisHelp;
using StackExchange.Redis;
using System.Collections.Concurrent;
using System.Data.Common;
using System.Diagnostics;
using System.Net.Sockets;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Xml;
using static Dm.net.buffer.ByteArrayBuffer;
using EnumHelper = GXRTBTC.PublicClassLibrary.Util.EnumHelper;

namespace GXRTBTC.WebReceiveService
{
    /// <summary>
    /// 后台服务
    /// 使用后台服务，让服务已启动就对Worker进行初始化
    /// </summary>
    public class MyBackgroundService : Microsoft.Extensions.Hosting.BackgroundService
    {
        /// <summary>
        /// 实例Worker这里引用只是为了让其初始化
        /// </summary>
        /// <param name="logger"></param>
        /// <param name="worker"></param>
        public MyBackgroundService(Worker worker) { }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            return Task.CompletedTask;
        }
    }

    /// <summary>
    /// 接收到的数据
    /// </summary>
    public class ReceiveData
    {
        public Func<ReceiveData, Object>? func { get; set; }
        public Object data { get; set; } = "";
        public string ip { get; set; } = "";
        public ushort port { get; set; } = 0;
        /// <summary>
        /// 0-数字信源，json
        /// 1-科江新远场，xml
        /// 2-新webapi接口，json
        /// 3-新webapi接口，json，补推
        /// </summary>
        public int type { get; set; } = 0;
        public DateTime dateTime { get; set; } = DateTime.Now;
    }

    /// <summary>
    /// 后台服务
    /// </summary>
    public class Worker
    {
        #region 参数
        /// <summary>
        /// 当前节点名称
        /// 组成：节点所在IP+容器ID
        /// </summary>
        //public string nodeName { get; set; } = TransForm.GetContainerAddress() + "_" + Environment.MachineName;
        public string nodeName { get; set; } = TransForm.GetHostIPOrName() + "_" + Environment.MachineName;
        /// <summary>
        /// 当前节点ID 即UUID
        /// </summary>
        //private string uuid { get; set; } = Guid.NewGuid().ToString("N");
        /// <summary>
        /// 心跳节点名称，保证唯一
        /// </summary>
        private string keySvr { get; set; } = "WebReceiveService";
        /// <summary>
        /// 日志
        /// </summary>
        public readonly ILogger mLogger;
        /// <summary>
        /// 配置
        /// </summary>
        private readonly IConfiguration mConfiguration;
        /// <summary>
        /// StackExchange.Redis
        /// </summary>
        public RedisHelper redisInit { get; set; }
        public RedisHelper redisYc { get; set; }
        public IBatch ycBatch { get; set; } = null;
        public IBatch svrBatch { get; set; } = null;
        public IBatch addrBatch { get; set; } = null;
        private RedisHelper redisGlobal { get; set; }
        private RedisHelper redisSTAT { get; set; }
        private RedisHelper redisHK { get; set; }
        private string redisConnString { get; set; } = "";
        /// <summary>
        /// 码流记录仪告警位
        /// </summary>
        private Dictionary<string, string> avAlarmCodes { get; set; } = new();
        private Dictionary<string, Dictionary<string, string>> avPdIds { get; set; } = new();

        /// <summary>
        /// 当前节点状态统计信息
        /// </summary>
        private Dictionary<string, Object> nodeStatusInfo { get; set; } = new Dictionary<string, object>() {
            { "Datetime", TransForm.GetFormatDatetime() },
            { "GetSum", 0 },
            { "GetRate", 0 },
            { "SkipNum", 0 },
            { "nodeNo",0 }
        };
        /// <summary>
        /// 副本数，默认只有自身
        /// </summary>
        private int nodeNum { get; set; } = 1;
        /// <summary>
        /// 当前副本编号
        /// </summary>
        private int nodeNo { get; set; } = 0;
        /// <summary>
        /// 对象锁
        /// </summary>
        public Object syncLock { get; set; } = new object();
        /// <summary>
        /// 接收数据总量
        /// </summary>
        private long numsRcv { get; set; } = 0;
        /// <summary>
        /// 接收、发送数量差，用于判断是否增加线程池数量
        /// </summary>
        private long gapNums { get; set; } = 0;
        /// <summary>
        /// 是否接收到数据
        /// </summary>
        private bool isRcvData { get; set; } = false;
        private bool isRcvXmlData { get; set; } = false;
        /// <summary>
        /// 跳过不处理的数量
        /// </summary>
        private long skipNum { get; set; } = 0;
        /// <summary>
        /// 锁对象
        /// </summary>
        private Object skipLock { get; set; } = new object();
        /// <summary>
        /// 最小线程池数量
        /// </summary>
        private int minPoolNum { get; set; } = 300;
        /// <summary>
        /// 最大线程池数量
        /// </summary>
        private int maxPoolNum { get; set; } = 500;
        /// <summary>
        /// 基础数据类型
        /// </summary>
        private Dictionary<string, string> basicType { get; set; } = new();

        /// <summary>
        /// 数据来源统计
        /// </summary>
        public Dictionary<string, long> dataStatistics { get; set; } = new Dictionary<string, long>();
        /// <summary>
        /// 信号源json数据接收缓存
        /// </summary>
        public BlockingCollection<ReceiveData> xyJsonCache { get; set; } = new BlockingCollection<ReceiveData>();
        /// <summary>
        /// 新远场xml数据接收缓存
        /// </summary>
        public BlockingCollection<ReceiveData> ycXmlCache { get; set; } = new BlockingCollection<ReceiveData>();
        /// <summary>
        /// 新监控系统WEB API接口协议
        /// </summary>
        public BlockingCollection<ReceiveData> jsonCache { get; set; } = new BlockingCollection<ReceiveData>();
        /// <summary>
        /// 数据缓存
        /// </summary>
        public BlockingCollection<PacketInfo> dataCacheToRedis { get; set; } = null; //Redis做消息队列
        public BlockingCollection<PacketInfo> xmlCacheToRedis { get; set; } = null; //Redis做消息队列
        public BlockingCollection<PacketInfo> jsonCacheToRedis { get; set; } = null; //Redis做消息队列
        /// <summary>
        /// ip:port缓存
        /// 用于更新台站、设备在线情况；
        /// </summary>
        public BlockingCollection<string> targetCache { get; set; } = null; //Redis做消息队列
        /// <summary>
        /// targetStation：台站ip=》台站编号
        /// targetPositionId：台站ip:port=》工位号
        /// </summary>
        private Dictionary<string, string> targetStation { get; set; } = new Dictionary<string, string>();
        private Dictionary<string, string> targetPositionId { get; set; } = new Dictionary<string, string>();
        /// <summary>
        /// 设备信息
        /// </summary>
        private Dictionary<string, DeviceInfo> deviceInfos { get; set; } = new Dictionary<string, DeviceInfo>();
        /// <summary>
        /// 所有台站
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> stationsInfo { get; set; } = new Dictionary<string, Dictionary<string, string>>();
        /// <summary>
        /// 台站，工位号对应设备信息
        /// </summary>
        private Dictionary<string, Dictionary<string, DeviceInfo>> stationPositionDeviceInfos { get; set; } = new Dictionary<string, Dictionary<string, DeviceInfo>>();
        /// <summary>
        /// 轮询下发信息
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> cmdSetInfo { get; set; } = new();
        private Dictionary<string, string> stationsDicts { get; set; } = new();
        /// <summary>
        /// 全局台站ID
        /// </summary>
        private string stationIdGlobal { get; set; } = "";
        /// <summary>
        /// web请求类
        /// </summary>
        private HttpClientUtil httpClientUtil { get; set; }
        /// <summary>
        /// 总平台web采集url
        /// </summary>
        private string webDataUrl { get; set; } = "https://10.170.1.234:8001/Receive";
        /// <summary>
        /// 台站信息缓存
        /// 台站名称为key
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> stationNames { get; set; } = new Dictionary<string, Dictionary<string, string>>();
        /// <summary>
        /// 台站信息缓存
        /// 台站ip地址为key
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> stationAddrs { get; set; } = new Dictionary<string, Dictionary<string, string>>();
        /// <summary>
        /// dtmb地址和台站地址一对一匹配数据
        /// 用于匹配新远场数据的台站
        /// </summary>
        //private Dictionary<string, string> dtmbStationMatch { get; set; } = new();
        /// <summary>
        /// 新远场IP判断
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> dtmbIp2StationId { get; set; } = new();
        /// <summary>
        /// 接收异常记录
        /// </summary>
        private Object exLock { get; set; } = new object();
        private Exception? exRcv { get; set; } = null;

        /// <summary>
        /// 配置集
        /// </summary>
        private Dictionary<string, string> paramInfo { get; set; } = new();
        #endregion

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="logger"></param>
        /// <param name="configuration"></param>
        public Worker(ILogger<Worker> logger, IConfiguration configuration)
        {
            this.mLogger = logger;
            this.mConfiguration = configuration;

            //给通用日志赋值
            CommonUtil.mLogger = logger;

            //初始化
            Init();
            mLogger.LogInformation("项目已启动");
        }

        #region 初始化
        public void Init()
        {
            //线程池配置
            ThreadPool.SetMinThreads(minPoolNum, minPoolNum);
            ThreadPool.SetMaxThreads(maxPoolNum, maxPoolNum);

            avAlarmCodes.TryAdd("515", "节目中断");
            avAlarmCodes.TryAdd("402", "视频丢失");
            avAlarmCodes.TryAdd("511", "视频加密");
            avAlarmCodes.TryAdd("403", "视频解码异常");
            avAlarmCodes.TryAdd("513", "视频马赛克");
            avAlarmCodes.TryAdd("502", "静帧");
            avAlarmCodes.TryAdd("503", "黑场");
            avAlarmCodes.TryAdd("505", "彩条");
            avAlarmCodes.TryAdd("506", "彩场");
            avAlarmCodes.TryAdd("404", "音频丢失");
            avAlarmCodes.TryAdd("512", "音频加密");
            avAlarmCodes.TryAdd("405", "音频解码异常");
            avAlarmCodes.TryAdd("510", "音量过低");
            avAlarmCodes.TryAdd("509", "音量过高");
            avAlarmCodes.TryAdd("504", "静音");

            //avPdIds
            avPdIds.TryAdd("6", new());
            avPdIds["6"].TryAdd("1", "CCTV1");
            avPdIds["6"].TryAdd("2", "CCTV2");
            avPdIds["6"].TryAdd("3", "CCTV4");
            avPdIds["6"].TryAdd("4", "CCTV10");
            avPdIds["6"].TryAdd("5", "CCTV12");
            avPdIds["6"].TryAdd("6", "CCTV13");
            avPdIds["6"].TryAdd("7", "CCTV14");
            avPdIds["6"].TryAdd("8", "CCTV15");

            avPdIds.TryAdd("7", new());
            avPdIds["7"].TryAdd("9", "CCTV9");
            avPdIds["7"].TryAdd("10", "CCTV11");
            avPdIds["7"].TryAdd("11", "CCTV17");
            avPdIds["7"].TryAdd("12", "CGTN");

            try
            {
                //redis初始化
#if DEBUG
                redisConnString = RsaEncrypt.DecryptFree(mConfiguration["RedisConnStringDebug"]);
#else
                redisConnString = RsaEncrypt.DecryptFree(mConfiguration["RedisConnString"]);
#endif
                redisInit = new RedisHelper(redisConnString, mLogger);
                redisYc = new RedisHelper(redisConnString, mLogger);
                redisGlobal = new RedisHelper(redisConnString, mLogger);
                redisSTAT = new RedisHelper(redisConnString, mLogger);
                redisHK = new RedisHelper(redisConnString, mLogger);

                ycBatch = redisYc.CreateBatch();
                svrBatch = redisYc.CreateBatch();
                addrBatch = redisYc.CreateBatch();
            }
            catch { }

            //数据缓存
            dataCacheToRedis = new BlockingCollection<PacketInfo>();
            xmlCacheToRedis = new BlockingCollection<PacketInfo>();
            jsonCacheToRedis = new BlockingCollection<PacketInfo>();
            targetCache = new BlockingCollection<string>();
            httpClientUtil = new HttpClientUtil(mLogger);

            //从缓存读取设置
            paramInfo = redisInit.HashGetAllDict<string>("paramInfo");

            //缓存信息更新
            InitCacheInfo();

            //基础数据类型，用于判断是否直接保存
            //basicType.TryAdd(typeof(sbyte).Name, "");
            //basicType.TryAdd(typeof(byte).Name, "");
            basicType.TryAdd(typeof(short).Name, "");
            basicType.TryAdd(typeof(ushort).Name, "");
            basicType.TryAdd(typeof(int).Name, "");
            basicType.TryAdd(typeof(uint).Name, "");
            basicType.TryAdd(typeof(long).Name, "");
            basicType.TryAdd(typeof(ulong).Name, "");
            basicType.TryAdd(typeof(float).Name, "");
            basicType.TryAdd(typeof(double).Name, "");
            basicType.TryAdd(typeof(char).Name, "");
            basicType.TryAdd(typeof(string).Name, "");
            basicType.TryAdd(typeof(bool).Name, "");

            //new Thread(Statistics).Start();     //统计
            new Thread(HeartBeatTask).Start();  //心跳

            new Thread((ThreadStart)(delegate { DeaRcvlData(xyJsonCache); })).Start();    //处理数字信源返回的json数据
            mLogger.LogInformation("启动数字信源数据处理线程");
            new Thread((ThreadStart)(delegate { DeaRcvlData(ycXmlCache); })).Start();   //处理新远场返回的xml数据
            mLogger.LogInformation("启动科江远场数据处理线程");
            new Thread((ThreadStart)(delegate { DeaRcvlData(jsonCache); })).Start();   //处理监控系统WEB API接口协议数据
            mLogger.LogInformation("启动监控系统WEB API接口协议数据处理线程");

            new Thread((ThreadStart)(delegate { DealDataToRedis(dataCacheToRedis); })).Start();   //缓存数据处理线程
            new Thread((ThreadStart)(delegate { DealDataToRedis(xmlCacheToRedis); })).Start();   //缓存数据处理线程
            new Thread((ThreadStart)(delegate { DealDataToRedis(jsonCacheToRedis); })).Start();   //缓存数据处理线程

            new Thread(GetFromSvr).Start();   //定时从总平台获取数据
            new Thread(PollingSetting).Start();   //根据访问情况从设备读取设置数据
            new Thread(UpdateOnlineStatus).Start();   //更新台站和设备在线信息数据
            new Thread(TimerTask).Start();      //定时监视判断是否增加线程池数量来处理数据
        }

        /// <summary>
        /// redis缓存信息初始化
        /// </summary>
        private void InitCacheInfo()
        {
            try
            {
                Dictionary<string, string> tmpTargetStation = redisInit.HashGetAllDict<string>("ipaddr_station");
                Dictionary<string, string> tmpTargetPositionId = redisInit.HashGetAllDict<string>("target_position");
                if (tmpTargetStation != null)
                {
                    lock (targetStation)
                    {
                        targetStation = tmpTargetStation;
                    }
                }
                if (tmpTargetPositionId != null)
                {
                    lock (targetPositionId)
                    {
                        targetPositionId = tmpTargetPositionId;
                    }
                }

                //台站工位号设备信息
                Dictionary<string, Dictionary<string, DeviceInfo>> tmpStationPositionDeviceInfos = redisInit.HashGetAllDict<Dictionary<string, DeviceInfo>>("station_position_deviceInfos");
                if (tmpStationPositionDeviceInfos != null && tmpStationPositionDeviceInfos.Count > 0)
                {
                    lock (stationPositionDeviceInfos)
                    {
                        stationPositionDeviceInfos = tmpStationPositionDeviceInfos;
                    }
                }

                Dictionary<string, Dictionary<string, string>> tmpCmdSetInfo = redisInit.HashGetAllDict<Dictionary<string, string>>("cmdSetInfo");
                cmdSetInfo = (tmpCmdSetInfo != null && tmpCmdSetInfo.Count > 0) ? tmpCmdSetInfo : cmdSetInfo;

                Dictionary<string, Dictionary<string, string>> tmpStationsInfo = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("stations");
                stationsInfo = (tmpStationsInfo != null && tmpStationsInfo.Count > 0) ? tmpStationsInfo : stationsInfo;
                stationsDicts = stationsInfo.Keys.ToDictionary(l => l, linkNums => "");

                //全局台站编号
                stationIdGlobal = (stationsInfo.Count > 0 ? stationsInfo.First().Key : stationIdGlobal);

                //从缓存读取设置
                paramInfo = redisInit.HashGetAllDict<string>("paramInfo");
                webDataUrl = paramInfo.GetValueOrDefault("WebReceiveDataUrl", webDataUrl);

                //台站信息缓存
                //台站名称为key
                Dictionary<string, Dictionary<string, string>> tmpStationNames = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("stationNames");
                stationNames = (tmpStationNames != null && tmpStationNames.Count > 0) ? tmpStationNames : stationNames;

                //台站信息缓存
                //台站ip地址为key
                Dictionary<string, Dictionary<string, string>> tmpStationAddrs = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("stationAddrs");
                stationAddrs = (tmpStationAddrs != null && tmpStationAddrs.Count > 0) ? tmpStationAddrs : stationAddrs;

                //deviceInfos
                var tmpDeviceInfos = redisGlobal.HashGetAllDict<DeviceInfo>("station_deviceInfos");
                List<string> pids = deviceInfos.Keys.ToList();
                foreach (string positionId in pids)
                {
                    if (!tmpDeviceInfos.ContainsKey(positionId) || deviceInfos[positionId].new_time > tmpDeviceInfos[positionId].new_time)
                    {
                        redisGlobal.HashSet("station_deviceInfos", positionId, deviceInfos[positionId]);
                    }
                }

                ////dtmb地址和台站地址一对一匹配数据
                //Dictionary<string, string> tmpDtmbStationMatch = redisInit.StringGet<Dictionary<string, string>>("dtmbStationMatch");
                //dtmbStationMatch = (tmpDtmbStationMatch != null && tmpDtmbStationMatch.Count > 0) ? tmpDtmbStationMatch : dtmbStationMatch;

                Dictionary<string, Dictionary<string, string>> tmpDtmbIp2StationId = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("dtmbIp2StationId");
                dtmbIp2StationId = (tmpDtmbIp2StationId != null && tmpDtmbIp2StationId.Count > 0) ? tmpDtmbIp2StationId : dtmbIp2StationId;
            }
            catch (Exception e)
            {
                mLogger.LogInformation("更新获取台站信息缓存失败，" + e);
            }
        }
        #endregion

        /// <summary>
        /// 设置访问统计
        /// </summary>
        /// <param name="httpContext"></param>
        public void SetStatistics(HttpContext HttpContext)
        {

        }

        #region 海康威视人脸识别回调
        /// <summary>
        /// 海康威视人脸识别回调
        /// </summary>
        /// <param name="faceAlarmData"></param>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public async Task<Object> HKFaceCallBack(string faceAlarmData, HttpContext httpContext)
        {
            var sw = Stopwatch.StartNew();
            try
            {
                mLogger.LogInformation("HKFaceCallBack,data=" + faceAlarmData);

                string key = "HKFaceCallBack_" + TransForm.GetFormatDatetime("yyyyMMdd");
                redisHK.HashSet(key, TransForm.GetFormatDatetime(), faceAlarmData);
                redisHK.KeyExpire(key, TimeSpan.FromDays(7));
            }
            catch (Exception e)
            {
                mLogger.LogError("HKFaceCallBack error msg:" + e.Message);
                return false;
            }
            return true;
        }
        #endregion

        #region 科江远场数据
        /// <summary>
        /// 接收到json数据，先写入缓存
        /// </summary>
        /// <param name="data"></param>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public Object DataFromXmlToCache(string data, HttpContext httpContext)
        {
            var sw = Stopwatch.StartNew();
            Dictionary<string, Object> rs = new() { { "return_code", 0 }, { "return_msg", "success" }, { "node_name", nodeName }, { "time", -1 }, { "current", TransForm.DateTimeFormat(DateTime.Now) } };
            try
            {
                string ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("::FFFF:", "");
                ushort port = (ushort)httpContext.Connection.RemotePort;
                ReceiveData receiveData = new ReceiveData()
                {
                    func = DealDataYC,
                    data = data,
                    ip = ip,
                    port = port,
                    type = 1
                };
                lock (ycXmlCache)
                {
                    ycXmlCache.Add(receiveData);
                }
            }
            catch (Exception e)
            {
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
            }
            finally
            {
                rs["time"] = sw.ElapsedMilliseconds;
            }

            return rs;
        }

        /// <summary>
        /// 科江远场数据
        /// 20240710临时调试接口
        /// </summary>
        /// <param name="data"></param>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public Object GetDataYC(string data, HttpContext httpContext)
        {
            var sw = Stopwatch.StartNew();
            Dictionary<string, Object> rs = new() { { "return_code", 0 }, { "return_msg", "success" }, { "node_name", nodeName }, { "time", -1 }, { "current", TransForm.DateTimeFormat(DateTime.Now) } };
            try
            {
                string ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("::FFFF:", "");
                ushort port = (ushort)httpContext.Connection.RemotePort;
                ReceiveData receiveData = new ReceiveData()
                {
                    func = DealDataYC,
                    data = data,
                    ip = ip,
                    port = port,
                    type = 1
                };

                rs = (Dictionary<string, Object>)receiveData.func(receiveData);
            }
            catch (Exception e)
            {
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
            }
            finally
            {
                rs["time"] = sw.ElapsedMilliseconds;
            }

            return rs;
        }

        /// <summary>
        /// 科江远场数据
        /// 20240710临时调试接口
        /// </summary>
        /// <param name="receiveData"></param>
        /// <returns></returns>
        public Object DealDataYC(ReceiveData receiveData)
        {
            var sw = Stopwatch.StartNew();
            Dictionary<string, Object> rs = new() { { "return_code", 0 }, { "return_msg", "success" }, { "node_name", nodeName }, { "time", -1 }, { "current", TransForm.DateTimeFormat(receiveData.dateTime) } };
            try
            {
                //先写一条，代表确认收到数据 Alarm
                string tmpMsgType = receiveData.data.ToString().Contains("Alarm") ? "AlarmAutoMsg" : "StatusAutoMsg";
                ycBatch.MHashSetAsync("DataYcTime", receiveData.ip + "_" + tmpMsgType + "_time", TransForm.DateTimeFormat(receiveData.dateTime));

                //组装
                PacketInfo packetInfo = new PacketInfo
                {
                    ip = receiveData.ip,
                    port = receiveData.port,
                    receivingTime = receiveData.dateTime,
                    receiveNode = nodeName + "_" + (nodeNo + 1).ToString(),
                    receiveNo = (nodeNo + 1).ToString(),
                    tipMsg = "HTTPXML",    //HTTPXML 标注数据来源及数据格式
                    dataType = "3",
                    reviceDatas = receiveData.data.ToString()
                };

                //帧体识别补充
                IdentifyXmlData(packetInfo);

                XmlDocument doc = new XmlDocument();
                //''是特殊字符，会导致转换失败，这里替换掉
                doc.LoadXml(receiveData.data.ToString().Replace("“", "\"").Replace("”", "\"").Replace("\u0013", ""));
                Dictionary<string, string> valueDict = GetDataFromXmlData(packetInfo, doc, out string msgType, out string macaddr, out Dictionary<string, Object> positionIds);
                packetInfo.header.msgType = msgType;
                //这里structData用来存储工位号信息
                if (positionIds != null && positionIds.Count > 0 && (packetInfo.positionId == null || "".Equals(packetInfo.positionId)))
                {
                    packetInfo.positionId = positionIds.Keys.ToList()[0].ToString();
                    packetInfo.stationCode = packetInfo.positionId?.Substring(0, packetInfo.positionId.IndexOf("_"));
                    if (stationsInfo.ContainsKey(packetInfo.stationCode ?? ""))
                    {
                        packetInfo.stationLevel = stationsInfo[packetInfo.stationCode ?? ""]["station_level"];
                    }

                    if ("yes".Equals(packetInfo.header.errorFlag))
                    {
                        packetInfo.tipMsg += "，台站信息从内容获取";
                    }
                    packetInfo.header.errorFlag = "no";
                }
                packetInfo.structData = positionIds;
                packetInfo.dictData = valueDict;
                packetInfo.seviceSN = macaddr;

                //赋值设备序列号
                CheckDeviceInfos(packetInfo, macaddr);

                //ycBatch.MHashSetAsync("DataYcData", receiveData.ip + "_" + msgType + "_data", receiveData.data);
                ycBatch.MHashSetAsync("DataYcData", receiveData.ip + "_" + msgType + "_" + packetInfo.stationCode, receiveData.data.ToString());
                //ycBatch.MHashSetAsync("DataYcTime", receiveData.ip + "_" + msgType + "_" + TransForm.DateTimeFormat(receiveData.dateTime, "HH"), TransForm.DateTimeFormat(receiveData.dateTime));
                //ycBatch.MHashSetAsync("DataYcTime", receiveData.ip + "_" + msgType + "_" + packetInfo.stationCode + "_" + TransForm.DateTimeFormat(receiveData.dateTime, "HH"), TransForm.DateTimeFormat(receiveData.dateTime));

                //正常的数据才写redis
                if (!"yes".Equals(packetInfo.header.errorFlag))
                {
                    lock (xmlCacheToRedis)
                    {
                        xmlCacheToRedis.Add(packetInfo);
                    }

                    lock (targetCache)
                    {
                        //targetCache.Add(receiveData.ip + "," + packetInfo.positionId + "," + JsonConvert.SerializeObject(packetInfo.receivingTime));
                        //多个台站的，轮询
                        if (positionIds != null && positionIds.Count > 0)
                        {
                            foreach (var positionId in positionIds.Keys)
                            {
                                targetCache.Add(receiveData.ip + "," + positionId + "," + JsonConvert.SerializeObject(packetInfo.receivingTime));
                            }
                        }
                    }
                }

                lock (syncLock)
                {
                    gapNums++;
                    numsRcv++;
                    if (!isRcvXmlData)
                    {
                        isRcvXmlData = true;
                        mLogger.LogInformation("接收到XML数据，开始处理");
                    }
                }
            }
            catch (Exception e)
            {
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
            }
            finally
            {
                rs["time"] = sw.ElapsedMilliseconds;
            }

            return rs;
        }

        /// <summary>
        /// xml
        /// 帧体识别补充
        /// </summary>
        /// <param name="packetInfo"></param>
        private void IdentifyXmlData(PacketInfo packetInfo)
        {
            //设置帧头
            SetXmlHeader(packetInfo);

            ////根据IP匹配台站
            //if (dtmbStationMatch.ContainsKey(packetInfo.ip) && stationAddrs.ContainsKey(dtmbStationMatch[packetInfo.ip]))
            //{
            //    string stationIp = dtmbStationMatch[packetInfo.ip];

            //    //得到台站信息
            //    packetInfo.stationCode = stationAddrs[stationIp]["station_id"];
            //    packetInfo.stationLevel = stationAddrs[stationIp]["station_level"];

            //    //组成工位号，新远场工位号后缀默认为3
            //    //工位号 台站编号+设备类型+工位编号
            //    packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.deviceType + "_3";
            //}
            ////有可能是本台站，如平果岜化山转播台
            //else if (dtmbStationMatch.ContainsValue(packetInfo.ip) && stationAddrs.ContainsKey(packetInfo.ip))
            //{
            //    string stationIp = packetInfo.ip;

            //    //得到台站信息
            //    packetInfo.stationCode = stationAddrs[stationIp]["station_id"];
            //    packetInfo.stationLevel = stationAddrs[stationIp]["station_level"];

            //    //组成工位号，新远场工位号后缀默认为3
            //    //工位号 台站编号+设备类型+工位编号
            //    packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.deviceType + "_3";

            //    packetInfo.tipMsg = "本台站远场数据在本台站，如有误，请核查！";
            //}
            //else
            //{
            //    packetInfo.positionId = "";

            //    packetInfo.header.errorFlag = "yes";
            //    packetInfo.tipMsg = "IP未知";
            //}

            if (dtmbIp2StationId.ContainsKey(packetInfo.ip))
            {
                //string stationId = dtmbIp2StationId[packetInfo.ip].Keys.ToList()[0];

                ////得到台站信息
                //packetInfo.stationCode = stationId;

                ////组成工位号，新远场工位号后缀默认为3
                ////工位号 台站编号+设备类型+工位编号
                //packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.deviceType + "_3";
            }
            else
            {
                packetInfo.header.errorFlag = "yes";
                packetInfo.tipMsg = "IP未知";
            }

            ////赋值设备序列号
            //DeviceInfo deviceInfo = GetDeviceInfo(packetInfo);
            //packetInfo.seviceSN = deviceInfo == null ? "" : deviceInfo.seviceSN;
        }

        /// <summary>
        /// 设置帧头
        /// </summary>
        /// <param name="packetInfo"></param>
        private void SetXmlHeader(PacketInfo packetInfo)
        {
            string msgType = "StatusAutoMsg";
            //msgType = "AlarmAutoMsg";
            if (packetInfo.reviceDatas.ToUpper().Contains("ALARM"))
            {
                msgType = "AlarmAutoMsg";
            }

            packetInfo.header = new HeaderData()
            {
                devaddr = packetInfo.port,
                msgType = msgType,
                device = "0x0902",
                errorFlag = "no"
            };
        }

        /// <summary>
        /// 解析xml文件
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <param name="doc"></param>
        /// <param name="msgType"></param>
        /// <param name="macaddr"></param>
        /// <param name="positionIds">用来存储所有的工位号</param>
        /// <returns></returns>
        private Dictionary<string, string> GetDataFromXmlData(PacketInfo packetInfo, XmlDocument doc, out string msgType, out string macaddr, out Dictionary<string, Object> positionIds)
        {
            Dictionary<string, string> valueDict = new();
            string positionId = packetInfo.positionId;
            positionIds = new Dictionary<string, Object>();
            if (packetInfo.positionId != null && !"".Equals(packetInfo.positionId))
            {
                positionIds.TryAdd(packetInfo.positionId, "");
            }
            msgType = "StatusAutoMsg";
            macaddr = "";
            try
            {
                //获得根节点
                XmlNode rootEle = doc.SelectSingleNode("Msg");
                if (rootEle != null && rootEle.HasChildNodes)
                {
                    bool isDtmbStatus = false;
                    bool isDtmbAlarm = false;
                    int fmIdx = 0;
                    int dtvIdx = 0;

                    XmlElement rootNode = (XmlElement)rootEle;//将节点转换为元素
                    ////状态和告警的不一致，暂时不用
                    //string srcCode = rootNode.GetAttribute("SrcCode");
                    //只有告警有
                    macaddr = rootNode.GetAttribute("MacAddr");

                    //20250410 根据与玉龙沟通，新增返回设备IP，根据ip最后一段，拼接成流媒体请求地址，如：
                    //192.168.1.229 22990
                    //192.168.1.228 22890
                    string IPadd = rootNode.GetAttribute("IPadd") ?? "";
                    //存储在memo里面
                    packetInfo.header.memo = IPadd;

                    //EqCode 设备地址
                    //string devAddr = rootNode.GetAttribute("EqCode");

                    foreach (XmlNode xn in rootEle.ChildNodes)
                    {
                        if (xn is XmlElement && xn.Name.Equals("Return"))
                        {
                            //DTMBParamP
                            XmlElement returnMsg = (XmlElement)xn;//将节点转换为元素
                            string xmlType = returnMsg.GetAttribute("Type");
                            string val = returnMsg.GetAttribute("Value");
                            if ("DTMBParamP".Equals(xmlType) && "0".Equals(val))
                            {
                                isDtmbStatus = true;
                                msgType = "StatusAutoMsg";
                            }
                            else if ("AlarmSearchPSet".Equals(xmlType) && "0".Equals(val))
                            {
                                isDtmbAlarm = true;
                                msgType = "AlarmAutoMsg";
                            }
                        }
                        //状态
                        else if (isDtmbStatus && xn is XmlElement && xn.Name.Equals("ReturnInfo") && xn.HasChildNodes)
                        {
                            XmlElement returnInfo = (XmlElement)xn;//将节点转换为元素
                            foreach (XmlNode xnParam in returnInfo.ChildNodes)
                            {
                                if (xnParam.HasChildNodes)
                                {
                                    XmlElement xnSets = (XmlElement)xnParam;//将节点转换为元素
                                    //频率格式为101.70MHz等，包含2位小数点，所以要替换掉得到真实频率
                                    string freq = xnSets.GetAttribute("Freq").ToUpper().Replace("0MHZ", "").Replace("MHZ", "").Replace("0KHZ", "").Replace("KHZ", "");
                                    //台站编号
                                    string pad = (xnSets.HasAttribute("Pad") && stationsDicts.ContainsKey(xnSets.GetAttribute("Pad"))) ? xnSets.GetAttribute("Pad") : "";
                                    string pid = (pad != null && !"".Equals(pad)) ? pad + "_" + packetInfo.deviceType + "_3" : "";

                                    if (!"".Equals(pid) && !positionIds.ContainsKey(pid))
                                    {
                                        positionIds.TryAdd(pid, "");
                                    }

                                    //直接在valueDict的key前面拼接上工位号
                                    pid = "DATA_" + pid + "#";

                                    //调频为FM，数字电视无Type参数
                                    string setType = xnSets.GetAttribute("Type");
                                    string programName = xnSets.GetAttribute("Name");
                                    int idx = fmIdx;
                                    //判断是否为数字电视，数字电视需要把频率转为频点
                                    //if (!xnSets.HasAttribute("Type") && "FM".Equals(setType))
                                    //{
                                    //    idx = dtvIdx;
                                    //    dtvIdx++;

                                    //    //转出频点，如21、28
                                    //    freq = TransForm.GetDTMBChannelNo(float.Parse(freq)).ToString();
                                    //}
                                    //else
                                    //{
                                    //    idx = fmIdx;
                                    //    fmIdx++;
                                    //}

                                    if (!xnSets.HasAttribute("Type"))
                                    {
                                        idx = dtvIdx;
                                        dtvIdx++;

                                        //转出频点，如21、28
                                        freq = TransForm.GetDTMBChannelNo(float.Parse(freq)).ToString();
                                    }
                                    else
                                    {
                                        if (!"FM".Equals(setType))
                                        {
                                            idx = dtvIdx;
                                            dtvIdx++;

                                            //转出频点，如21、28
                                            freq = TransForm.GetDTMBChannelNo(float.Parse(freq)).ToString();
                                        }
                                        else
                                        {
                                            idx = fmIdx;
                                            fmIdx++;
                                        }
                                    }

                                    //string key = "FREQ_" + xnSets.Name + "_" + xn.Name + "_" + msgType + "_" + idx;
                                    string key = idx + "_" + xnSets.Name + "_FREQ";
                                    valueDict.TryAdd(pid + key.ToUpper(), freq);

                                    if (!"".Equals(programName))
                                    {
                                        key = idx + "_" + xnSets.Name + "_NAME";
                                        valueDict.TryAdd(pid + key.ToUpper(), programName);
                                    }

                                    foreach (XmlNode xnSet in xnSets.ChildNodes)
                                    {
                                        XmlElement setDetails = (XmlElement)xnSet;
                                        string name = setDetails.GetAttribute("Desc");
                                        string val = setDetails.GetAttribute("Value");

                                        key = idx + "_" + xnSets.Name + "_" + name;
                                        valueDict.TryAdd(pid + key.ToUpper(), val);
                                    }
                                }
                            }
                        }
                        //告警
                        else if (isDtmbAlarm && xn is XmlElement && xn.Name.Equals("ReturnInfo") && xn.HasChildNodes)
                        {
                            XmlElement returnInfo = (XmlElement)xn;//将节点转换为元素
                            //1、防止重复key出错
                            //2、存在这种情况时，取最新值和时间
                            Dictionary<string, DateTime> tmpKeys = new();
                            foreach (XmlNode xnParam in returnInfo.ChildNodes)
                            {
                                if (xnParam.HasChildNodes)
                                {
                                    XmlElement xnSets = (XmlElement)xnParam;//将节点转换为元素
                                    //频率格式为101.70MHz等，包含2位小数点，所以要替换掉得到真实频率
                                    string freq = xnSets.GetAttribute("Freq").ToUpper().Replace("0MHZ", "").Replace("MHZ", "");
                                    //台站编号
                                    string pad = (xnSets.HasAttribute("Pad") && stationsDicts.ContainsKey(xnSets.GetAttribute("Pad"))) ? xnSets.GetAttribute("Pad") : "";
                                    string pid = (pad != null && !"".Equals(pad)) ? pad + "_" + packetInfo.deviceType + "_3" : "";
                                    if (!"".Equals(pid) && !positionIds.ContainsKey(pid))
                                    {
                                        positionIds.TryAdd(pid, "");
                                    }

                                    //直接在valueDict的key前面拼接上工位号
                                    pid = "DATA_" + pid + "#";

                                    //自定义节目类型 0-调频 1-地数
                                    int jmType = 0;
                                    string vpid = xnSets.GetAttribute("VideoPID") ?? "0";
                                    if (vpid != null && !"".Equals(vpid) && TransForm.IsInt(vpid) && int.Parse(vpid) > 0)
                                    {
                                        jmType = 1;
                                    }

                                    //判断是否为数字电视，数字电视需要把频率转为频点
                                    if (jmType == 1)
                                    {
                                        //转出频点，如21、28
                                        freq = TransForm.GetDTMBChannelNo(float.Parse(freq)).ToString();
                                    }

                                    string svrid = xnSets.GetAttribute("ServiceID") ?? "0";
                                    string keySvrId = (freq + "_" + jmType + "_ALARM_SVRID").ToUpper();
                                    if (!valueDict.ContainsKey(keySvrId))
                                    {
                                        valueDict.TryAdd(pid + keySvrId, svrid);
                                    }

                                    foreach (XmlNode xnSet in xnSets.ChildNodes)
                                    {
                                        XmlElement setDetails = (XmlElement)xnSet;
                                        string tp = setDetails.GetAttribute("Type");
                                        string val = setDetails.GetAttribute("Value");
                                        string desc = setDetails.GetAttribute("Desc");
                                        string tm = setDetails.GetAttribute("Time");
                                        DateTime dt = TransForm.strTodateTime(tm);

                                        string keyVal = (tp + "_" + freq + "_" + jmType + "_ALARM_VALUE").ToUpper();
                                        string keyDesc = (tp + "_" + freq + "_" + jmType + "_ALARM_DESC").ToUpper();
                                        string keyTm = (tp + "_" + freq + "_" + jmType + "_ALARM_TIME").ToUpper();
                                        //记录最新时间的key
                                        string keyNow = keyVal + "_NEWTIME";
                                        if (!tmpKeys.ContainsKey(keyTm))
                                        {
                                            //值
                                            valueDict.TryAdd(pid + keyVal, val);
                                            valueDict.TryAdd(pid + keyDesc, desc);
                                            //时间
                                            valueDict.TryAdd(pid + keyTm, tm);

                                            //针对每一个告警，记录一个接收时间，用于判断告警是否结束
                                            valueDict.TryAdd(pid + keyNow, TransForm.DateTimeFormat(packetInfo.receivingTime));

                                            tmpKeys.TryAdd(keyTm, dt);
                                        }
                                        //如果key存在，则取时间最近的一条
                                        else if (dt > tmpKeys[keyTm])
                                        {
                                            //值
                                            valueDict[pid + keyVal] = val;
                                            valueDict[pid + keyDesc] = desc;
                                            //时间
                                            valueDict[pid + keyTm] = tm;

                                            //针对每一个告警，记录一个接收时间，用于判断告警是否结束
                                            valueDict[pid + keyNow] = TransForm.DateTimeFormat(packetInfo.receivingTime);

                                            tmpKeys[keyTm] = dt;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
            catch { }

            return valueDict;
        }
        #endregion

        #region 监控系统WEB API接口
        /// <summary>
        /// 针对websocket的token验证
        /// 预留，暂无用
        /// </summary>
        /// <param name="webApiData"></param>
        /// <returns></returns>
        public bool GetValidToken(IWebApiData webApiData)
        {
            bool rs = true;
            try
            {
                //var authHeader = httpContext.Request.Headers["Authorization"];
                //var token = authHeader.FirstOrDefault()?.Split(" ").Last(); // 假设使用Bearer token
                //rs = !string.IsNullOrEmpty(token);

                //获取token，进行验证
                //string token = webApiData.Data.ToString();
            }
            catch
            {
                rs = false;
            }

            return rs;
        }

        /// <summary>
        /// 参数检查
        /// </summary>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public WebParamData CheckParam(HttpContext httpContext)
        {
            WebParamData webParamData = new WebParamData();
            string msg = "";

            try
            {
                //路径参数
                string apiver = (httpContext.Request.RouteValues["apiver"] ?? httpContext.Request.Query["apiver"]).ToString();
                string level = (httpContext.Request.RouteValues["level"] ?? httpContext.Request.Query["level"]).ToString();
                string positionid = (httpContext.Request.RouteValues["positionid"] ?? httpContext.Request.Query["positionid"]).ToString();
                string sn = (httpContext.Request.RouteValues["sn"] ?? httpContext.Request.Query["sn"]).ToString();

                if (apiver == null || "".Equals(apiver))
                {
                    msg += "，API协议版本apiver不能为空";
                }

                if (level == null || "".Equals(level))
                {
                    msg += "，级别level（SYSTEM(系统级) DEVICE(设备级)）不能为空";
                }

                if (positionid == null || "".Equals(positionid))
                {
                    msg += "，位置编号（或工位号）positionid不能为空";
                }

                else if (positionid.Split("_").Length < 3)
                {
                    msg += "，位置编号（或工位号）positionid异常，参考格式xx_xxx_xxx";
                }

                if (sn == null || "".Equals(sn))
                {
                    msg += "，系统/设备序列号sn不能为空";
                }

                webParamData.apiver = apiver;
                webParamData.level = level;
                webParamData.positionid = positionid;
                webParamData.sn = sn;
            }
            catch (Exception e)
            {
                msg += e.Message;
            }
            finally
            {
                webParamData.nowmal = string.IsNullOrEmpty(msg);
                webParamData.msg = webParamData.nowmal ? msg : msg.Substring(1);
            }

            return webParamData;
        }

        /// <summary>
        /// 检查请求参数
        /// </summary>
        /// <param name="data"></param>
        /// <param name="webApiDataRes"></param>
        /// <returns></returns>
        public bool CheckData(string data, WebApiDataFull webApiDataFull)
        {
            bool rs = false;
            string msg = "";
            try
            {
                IWebApiData webApiData = null;
                Dictionary<string, Object> dictData = null;
                if (data != null && !"".Equals(data.ToString()) && ValidJson.IsJson(data.ToString()))
                {
                    dictData = JsonConvert.DeserializeObject<Dictionary<string, Object>>(data.ToString());
                    if (webApiDataFull.IsWebSocket)
                    {
                        webApiData = JsonConvert.DeserializeObject<WebApiDataReqWS>(data.ToString());
                    }
                    else
                    {
                        webApiData = JsonConvert.DeserializeObject<WebApiDataReq>(data.ToString());
                    }

                    if (webApiData == null)
                    {
                        msg += "，数据异常";
                    }
                }
                else
                {
                    msg += "，数据异常";
                }

                //数据必须参数检查
                if (dictData != null && dictData.Count > 0)
                {
                    Dictionary<string, string> tmpDict = dictData.ToDictionary(d => d.Key.ToUpper(), d => "");
                    if (!tmpDict.ContainsKey("REQUESTID"))
                    {
                        msg += "，数据不完整，缺少RequestId（请求编号）参数";
                    }
                    if (!tmpDict.ContainsKey("COMMAND"))
                    {
                        msg += "，数据不完整，缺少Command（命令）参数";
                    }
                    if (!tmpDict.ContainsKey("VERSION"))
                    {
                        msg += "，数据不完整，缺少Version（设备或系统通信协议版本）参数";
                    }
                    if (!tmpDict.ContainsKey("DATA"))
                    {
                        msg += "，数据不完整，缺少Data数据";
                    }
                    if (!tmpDict.ContainsKey("TIMESTAMP"))
                    {
                        msg += "，数据不完整，缺少Timestamp（时间戳）参数";
                    }
                    if (webApiDataFull.IsWebSocket && !tmpDict.ContainsKey("TYPE"))
                    {
                        msg += "，数据不完整，缺少Type（消息类型）参数";
                    }
                }

                webApiDataFull.dictData = dictData;
                webApiDataFull.ContentReq = webApiData;

                //消息类型判断
                if (webApiDataFull.IsWebSocket && webApiData != null && !EnumHelper.HasValue<WebApiTypeCodeWS>(((WebApiDataReqWS)webApiDataFull.ContentReq).Type))
                {
                    msg += "，数据不完整，Type（消息类型）异常";
                }

                //工位号格式处理，确保格式统一
                webApiDataFull.webParamData.positionid = webApiDataFull.webParamData.positionid.Replace("_0X", "_0x");
                rs = string.IsNullOrEmpty(msg);
                if (webApiDataFull.IsWebSocket)
                {
                    ((WebApiDataResWS)webApiDataFull.ContentRes).Code = rs ? WebApiStatusCode.code_200 : WebApiStatusCode.code_400;
                }
                else
                {
                    ((WebApiDataRes)webApiDataFull.ContentRes).Code = rs ? WebApiStatusCode.code_200 : WebApiStatusCode.code_400;
                }
            }
            catch (Exception e)
            {
                rs = false;
                //数据处理从错误
                msg += "，" + e.Message;
                if (webApiDataFull.IsWebSocket)
                {
                    ((WebApiDataResWS)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5504;
                }
                else
                {
                    ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5504;
                }
            }
            finally
            {
                try
                {
                    if (webApiDataFull.IsWebSocket)
                    {
                        ((WebApiDataResWS)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataResWS)webApiDataFull.ContentRes).Code) + msg;
                    }
                    else
                    {
                        ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + msg;
                    }

                    if (webApiDataFull.ContentReq != null)
                    {
                        webApiDataFull.ContentRes.RequestId = webApiDataFull.ContentReq.RequestId;
                        webApiDataFull.ContentRes.Command = webApiDataFull.ContentReq.Command;
                        webApiDataFull.ContentRes.Version = webApiDataFull.ContentReq.Version;
                        if (webApiDataFull.IsWebSocket)
                        {
                            if (rs)
                            {
                                ((WebApiDataResWS)webApiDataFull.ContentRes).Type = ((WebApiDataReqWS)webApiDataFull.ContentReq).Type.ToLower();
                            }
                            else
                            {
                                ((WebApiDataResWS)webApiDataFull.ContentRes).Type = WebApiTypeCodeWS.error.ToString();
                            }
                        }
                    }
                }
                catch
                {
                    rs = false;
                    //服务异常失败
                    if (webApiDataFull.IsWebSocket)
                    {
                        ((WebApiDataResWS)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5503;
                        ((WebApiDataResWS)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataResWS)webApiDataFull.ContentRes).Code);
                    }
                    else
                    {
                        ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5503;
                        ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code);
                    }
                }
            }

            return rs;
        }
        #region http/websocket
        /// <summary>
        /// 监控系统WEB API接口
        /// json格式
        /// </summary>
        /// <param name="webApiDataFull"></param>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public Object GetDataJson(WebApiDataFull webApiDataFull, HttpContext httpContext)
        {
            var sw = Stopwatch.StartNew();
            bool rs = false;
            try
            {
                string ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("::FFFF:", "");
                ushort port = (ushort)httpContext.Connection.RemotePort;

                //固定只处理某些命令
                //这里暂时只处理主动上报status和报警alarm的数据
                if (webApiDataFull.ContentReq.Command.ToUpper().Equals("STATUS") ||
                    webApiDataFull.ContentReq.Command.ToUpper().Equals("ALARM")
                    )
                {
                    ReceiveData receiveData = new ReceiveData()
                    {
                        func = GetDataJson,
                        data = webApiDataFull,
                        ip = ip,
                        port = port,
                        type = 2
                    };
                    lock (jsonCache)
                    {
                        jsonCache.Add(receiveData);
                    }
                    rs = true;
                }
                //状态、告警历史数据
                else if (webApiDataFull.ContentReq.Command.ToUpper().Equals("STATUS_HIS") || webApiDataFull.ContentReq.Command.ToUpper().Equals("ALARM_HIS"))
                {
                    //这部分数据只会给到历史数据服务用，存储到数据库
                    ReceiveData receiveData = new ReceiveData()
                    {
                        func = GetDataJson,
                        data = webApiDataFull,
                        ip = ip,
                        port = port,
                        type = 3
                    };
                    lock (jsonCache)
                    {
                        jsonCache.Add(receiveData);
                    }
                    rs = true;
                }
                //配置读取，直接返回
                else if (webApiDataFull.ContentReq.Command.ToUpper().EqualIgnoreCase("getdata"))
                {
                    rs = GetDataFromDev(webApiDataFull);
                    //mLogger.LogInformation(JsonConvert.SerializeObject(webApiDataFull));

                    ReceiveData receiveData = new ReceiveData()
                    {
                        func = GetDataJson,
                        data = webApiDataFull,
                        ip = ip,
                        port = port,
                        type = 2
                    };
                    lock (jsonCache)
                    {
                        jsonCache.Add(receiveData);
                    }
                }
                //操作下发，固定用operate
                //往主控、设备下发操作
                else if (webApiDataFull.ContentReq.Command.ToUpper().EqualIgnoreCase("operate"))
                {
                    rs = GetDataFromDev(webApiDataFull);
                    mLogger.LogInformation(JsonConvert.SerializeObject(webApiDataFull));
                }
                else
                {
                    ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_404;
                    ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "，无法响应命令【" + webApiDataFull.ContentReq.Command + "】";
                }
            }
            catch (Exception e)
            {
                ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5503;
                ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "," + e.Message;
            }

            return rs;
        }

        /// <summary>
        /// 监控系统WEB API接口协议
        /// 数据处理
        /// </summary>
        /// <param name="receiveData"></param>
        /// <returns></returns>
        public Object GetDataJson(ReceiveData receiveData)
        {
            var sw = Stopwatch.StartNew();
            Dictionary<string, Object> rs = new() { { "return_code", 0 }, { "return_msg", "success" }, { "node_name", nodeName }, { "time", -1 }, { "current", TransForm.DateTimeFormat(receiveData.dateTime) } };

            try
            {
                WebApiDataFull webApiDataFull = (WebApiDataFull)receiveData.data;
                //是否补推数据
                bool rePush = receiveData.type == 3;

                //组装
                PacketInfo packetInfo = new PacketInfo
                {
                    ip = receiveData.ip,
                    port = receiveData.port,
                    receivingTime = receiveData.dateTime,
                    receiveNode = nodeName + "_" + (nodeNo + 1).ToString(),
                    receiveNo = (nodeNo + 1).ToString(),
                    //JSONDATA 标注数据来源及数据格式
                    tipMsg = "JSONDATA",
                    //6是新webapi协议json补推的数据，其它是新webapi协议json
                    dataType = rePush ? "6" : (webApiDataFull.IsWebSocket ? "5" : "4"),
                    structData = webApiDataFull.dictData
                };

                packetInfo.dictData = GetDataFromJson(new Dictionary<string, string>(), packetInfo.structData);
                //数据格式处理，避免不同COMMAND数据相互覆盖
                if (packetInfo.dictData.ContainsKey("COMMAND"))
                {
                    string command = packetInfo.dictData["COMMAND"].ToUpper();
                    packetInfo.dictData = packetInfo.dictData.ToDictionary(d => d.Key.StartsWith("DATA#") ? d.Key : (d.Key + "_" + command), d => d.Value);
                }

                //帧体识别补充
                IdentifyJsonData(packetInfo);

                string positionId = webApiDataFull.webParamData.positionid.ToUpper().Replace("_0X", "_0x");
                packetInfo.positionId = positionId;
                packetInfo.stationCode = packetInfo.positionId.Split("_")[0];
                packetInfo.deviceType = packetInfo.positionId.Split("_")[1];
                if (stationsInfo.ContainsKey(packetInfo.stationCode ?? ""))
                {
                    packetInfo.stationLevel = stationsInfo[packetInfo.stationCode ?? ""]["station_level"];
                }
                packetInfo.seviceSN = webApiDataFull.webParamData.sn;

                //非补推的数据菜进行序列号处理
                if (!rePush)
                {
                    //赋值设备序列号
                    CheckDeviceInfos(packetInfo, packetInfo.seviceSN);
                }

                //正常的数据才写redis
                if (!"yes".Equals(packetInfo.header.errorFlag))
                {
                    lock (jsonCacheToRedis)
                    {
                        jsonCacheToRedis.Add(packetInfo);
                    }

                    lock (targetCache)
                    {
                        targetCache.Add(receiveData.ip + "," + packetInfo.positionId + "," + JsonConvert.SerializeObject(packetInfo.receivingTime));
                    }
                }

                lock (syncLock)
                {
                    gapNums++;
                    numsRcv++;
                    if (!isRcvXmlData)
                    {
                        isRcvXmlData = true;
                        mLogger.LogInformation("接收到监控系统WEBAPI新接口数据，开始处理");
                    }
                }
            }
            catch (Exception e)
            {
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
            }
            finally
            {
                rs["time"] = sw.ElapsedMilliseconds;
            }

            return rs;
        }

        /// <summary>
        /// 帧体识别补充
        /// </summary>
        /// <param name="packetInfo"></param>
        private void IdentifyJsonData(PacketInfo packetInfo)
        {
            //设置帧头
            SetJsonHeader(packetInfo);
        }

        /// <summary>
        /// 设置帧头
        /// </summary>
        /// <param name="packetInfo"></param>
        private void SetJsonHeader(PacketInfo packetInfo)
        {
            string msgType = "StatusAutoMsg";
            packetInfo.header = new HeaderData()
            {
                devaddr = packetInfo.port,
                msgType = msgType
            };
        }
        #endregion
        #endregion

        #region 数据处理
        /// <summary>
        /// 接收到json数据，先写入缓存
        /// </summary>
        /// <param name="data"></param>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public bool DataFromJsonToCache(string data, HttpContext httpContext)
        {
            bool rs = true;
            try
            {
                string ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("::FFFF:", "");
                ushort port = (ushort)httpContext.Connection.RemotePort;
                ReceiveData receiveData = new ReceiveData()
                {
                    func = DealDataFromJson,
                    data = data,
                    ip = ip,
                    port = port,
                    type = 0
                };

                lock (xyJsonCache)
                {
                    xyJsonCache.Add(receiveData);
                }
            }
            catch
            {
                rs = false;
            }

            return rs;
        }

        /// <summary>
        /// 接收到数据
        /// </summary>
        /// <param name="data"></param>
        /// <param name="httpContent"></param>
        /// <returns></returns>
        public bool GetDataFromJson(string data, HttpContext httpContext)
        {
            bool rs = true;
            try
            {
                string ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("::FFFF:", "");
                ushort port = (ushort)httpContext.Connection.RemotePort;
                ReceiveData receiveData = new ReceiveData()
                {
                    func = DealDataFromJson,
                    data = data,
                    ip = ip,
                    port = port,
                    type = 0
                };

                Dictionary<string, Object> result = (Dictionary<string, Object>)receiveData.func(receiveData);
                rs = int.Parse(result["return_code"].ToString()) == 0;
            }
            catch
            {
                rs = false;
            }

            return rs;
        }

        /// <summary>
        /// 接收到数据
        /// </summary>
        /// <param name="receiveData"></param>
        /// <returns></returns>
        public Object DealDataFromJson(ReceiveData receiveData)
        {
            Dictionary<string, Object> dataDict;
            Dictionary<string, Object> rs = new() { { "return_code", 0 }, { "return_msg", "success" }, { "node_name", nodeName }, { "time", -1 } };

            try
            {
                dataDict = JsonConvert.DeserializeObject<Dictionary<string, object>>(receiveData.data.ToString());
            }
            catch (Exception e)
            {
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
                return rs;
            }

            try
            {
                //如果堆积超过阈值（比如2千）条数据，就跳过
                if (gapNums > 2000)
                {
                    lock (skipLock)
                    {
                        skipNum++;
                    }
                    return true;
                }

                string from = receiveData.ip + ":" + receiveData.port;
                DateTime dtRcv = receiveData.dateTime;

                //统计
                lock (dataStatistics)
                {
                    if (dataStatistics.ContainsKey(from))
                    {
                        dataStatistics[from] += 1;
                    }
                    else
                    {
                        dataStatistics.TryAdd(from, 1);
                    }
                }

                //组装
                PacketInfo packetInfo = new PacketInfo
                {
                    ip = receiveData.ip,
                    port = receiveData.port,
                    receivingTime = dtRcv,
                    receiveNode = nodeName + "_" + (nodeNo + 1).ToString(),
                    receiveNo = (nodeNo + 1).ToString(),
                    tipMsg = "HTTPJSON",    //HTTPJSON 标注数据来源及数据格式
                    dataType = "1",
                    structData = dataDict
                };

                //帧体识别补充
                IdentifyData(packetInfo);
                //赋值设备序列号
                if (dataDict.ContainsKey("DevSN") && dataDict["DevSN"] != null)
                {
                    DeviceInfo deviceInfo = GetDeviceInfo(packetInfo, dataDict["DevSN"].ToString());
                    packetInfo.seviceSN = deviceInfo == null ? "" : deviceInfo.seviceSN;
                }
                else
                {
                    DeviceInfo deviceInfo = GetDeviceInfo(packetInfo);
                    packetInfo.seviceSN = deviceInfo == null ? "" : deviceInfo.seviceSN;
                }

                lock (dataCacheToRedis)
                {
                    dataCacheToRedis.Add(packetInfo);
                }
                lock (targetCache)
                {
                    targetCache.Add(receiveData.ip + "," + packetInfo.positionId + "," + JsonConvert.SerializeObject(packetInfo.receivingTime));
                }

                lock (syncLock)
                {
                    gapNums++;
                    numsRcv++;
                    if (!isRcvData)
                    {
                        isRcvData = true;
                        mLogger.LogInformation("接收到数据，开始处理");
                    }
                }
            }
            catch (Exception e)
            {
                lock (exLock)
                {
                    exRcv = e;
                }
                rs["return_code"] = -1;
                rs["return_msg"] = e.Message;
            }

            return rs;
        }

        /// <summary>
        /// 帧体识别补充
        /// </summary>
        /// <param name="packetInfo"></param>
        private void IdentifyData(PacketInfo packetInfo)
        {
            //设置帧头
            SetHeader(packetInfo);

            //deviceType设备带有
            string? deviceType = packetInfo.structData.ContainsKey("DevType") ? packetInfo.structData["DevType"].ToString() : "";
            if (deviceType != null)
            {
                if (deviceType.ToUpper().StartsWith("0X"))
                {
                    packetInfo.deviceType = deviceType ?? "";
                }
                else if (TransForm.IsInt(deviceType))
                {
                    packetInfo.deviceType = TransForm.IntToHex(int.Parse(deviceType));
                }
                else
                {
                    packetInfo.header.errorFlag = "yes";
                }
            }
            else
            {
                packetInfo.header.errorFlag = "yes";
            }

            //configVer写入到头信息的软件版本号钟
            string? configVer = packetInfo.structData.ContainsKey("configVer") ? packetInfo.structData["configVer"].ToString() : "";
            packetInfo.header.software = configVer ?? "";

            //将DevSN写到reviceDatas，反正空着也是空着
            string? DevSN = packetInfo.structData.ContainsKey("DevSN") ? packetInfo.structData["DevSN"].ToString() : "";
            packetInfo.reviceDatas = DevSN ?? "";

            //台站名称
            string? stationName = packetInfo.structData.ContainsKey("station_name") ? packetInfo.structData["station_name"].ToString() : "";

            //根据IP匹配台站
            if (stationAddrs.ContainsKey(packetInfo.ip))
            {
                //得到台站信息
                packetInfo.stationCode = stationAddrs[packetInfo.ip]["station_id"];
                packetInfo.stationLevel = stationAddrs[packetInfo.ip]["station_level"];

                //组成工位号，工位号后缀默认为1
                //工位号 台站编号+设备类型+工位编号
                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.deviceType + "_1";
            }
            else if (stationName != null && !"".Equals(stationName) && stationNames.ContainsKey(stationName))
            {

#if DEBUG
                //debug时
                packetInfo.tipMsg = "";
                packetInfo.ip = stationNames[stationName]["ip_addr"];
#else
                packetInfo.tipMsg = "IP未知，判断是台站【" + stationName + "】的数据，请核对";
                packetInfo.header.errorFlag = "yes";
#endif

                //得到台站信息
                packetInfo.stationCode = stationNames[stationName]["station_id"];
                packetInfo.stationLevel = stationNames[stationName]["station_level"];

                //组成工位号，工位号后缀默认为1
                //工位号 台站编号+设备类型+工位编号
                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.deviceType + "_1";
            }
            else
            {
                packetInfo.header.errorFlag = "yes";
                if (stationName != null && !"".Equals(stationName))
                {
                    packetInfo.tipMsg = "IP未知，可能是台站【" + stationName + "】的数据";
                }
                else
                {
                    packetInfo.tipMsg = "IP未知";
                }
            }

            ////赋值设备序列号
            //DeviceInfo deviceInfo = GetDeviceInfo(packetInfo);
            //packetInfo.seviceSN = deviceInfo == null ? "" : deviceInfo.seviceSN;
        }

        /// <summary>
        /// 设置帧头
        /// </summary>
        /// <param name="packetInfo"></param>
        private void SetHeader(PacketInfo packetInfo)
        {
            string msgType = "StatusAutoMsg";
            //report和alarm report是主动上报状态 alarm是告警
            string? type = packetInfo.structData.ContainsKey("type") ? packetInfo.structData["type"].ToString() : "";
            if (type != null && !"".Equals(type) && !type.ToUpper().Equals("REPORT"))
            {
                msgType = "AlarmAutoMsg";
            }

            packetInfo.header = new HeaderData()
            {
                devaddr = packetInfo.port,
                msgType = msgType,
                errorFlag = packetInfo == null ? "yes" : "no"
            };
        }

        /// <summary>
        /// 缓存数据处理线程
        /// </summary>
        private void DealDataToRedis(BlockingCollection<PacketInfo> dataCache)
        {
            RedisHelper redisTo = new RedisHelper(redisConnString, mLogger);
            RedisHelper redisToXml = new RedisHelper(redisConnString, mLogger);
            IBatch batch = redisTo.CreateBatch();
            IBatch batchXml = redisToXml.CreateBatch();
            while (true)
            {
                try
                {
                    while (dataCache.Count > 0)
                    {
                        PacketInfo packetInfo;
                        while (dataCache.Count > 0)
                        {
                            lock (dataCache)
                            {
                                packetInfo = dataCache.Take();
                            }

                            /// 0-普通数据，需要配置解析配置数据
                            /// 1-json数据，收到的就是json数据
                            /// 2-AI地数图像识别
                            /// 3-新远场xml数据
                            /// 4-从https获取的新监控系统WEB API接口协议数据
                            /// 5-从websocket获取的新监控系统WEB API接口协议数据
                            /// 6-新监控系统WEB API接口协议补推的数据，数据只用于历史数据服务，最终存储到历史表
                            if ("1".Equals(packetInfo.dataType))
                            {
                                DataCacheJson(packetInfo, batch, redisTo);
                            }
                            else if ("3".Equals(packetInfo.dataType))
                            {
                                DataCacheXml(packetInfo, batchXml, redisToXml);
                            }
                            else if ("4".Equals(packetInfo.dataType) || "5".Equals(packetInfo.dataType) || "6".Equals(packetInfo.dataType))
                            {
                                DataCacheNormalJson(packetInfo, batchXml, redisToXml);
                            }

                            lock (syncLock)
                            {
                                gapNums--;
                            }
                        }
                    }
                }
                catch { }
                finally
                {
                    Thread.Sleep(5);
                }
            }
        }
        #endregion

        #region 数据写缓存
        /// <summary>
        /// 缓存json数据
        /// 新标准格式
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <param name="batch"></param>
        /// <param name="redis"></param>
        private void DataCacheNormalJson(PacketInfo packetInfo, IBatch batch, RedisHelper redis)
        {
            try
            {
                //补推的数据，特殊处理
                if ("6".Equals(packetInfo.dataType))
                {
                    //还没考虑好，暂留空
                }
                else
                { 
                    //判断最新时间是否大于redis保存的时间 大于则返回，否则会造成数据跳动
                    //新增msgType用于判断
                    string redisKey = "newest_time_cache_" + packetInfo.header.msgType + "_" + packetInfo.deviceType;
                    DateTime newestSetTime = redis.StringGet<DateTime>(redisKey);
                    if ((newestSetTime >= packetInfo.receivingTime))
                    {
                        return;
                    }
                    redis.StringSet(redisKey, packetInfo.receivingTime);

                    //最新时间也写入
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "DATA_FROM", packetInfo.ip);
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "receiveNode", packetInfo.receiveNode);
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "dataType", packetInfo.dataType);
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "json_flag", 1);
                    foreach (string keyName in packetInfo.dictData.Keys)
                    {
                        batch.MHashSetAsync("DATA_" + packetInfo.positionId, keyName, packetInfo.dictData[keyName]);
                    }
                    batch.MExecute();
                }
            }
            catch (Exception e)
            {
                mLogger.LogError("JSON数据缓存失败！ " + e);
            }
        }

        /// <summary>
        /// json数据处理
        /// 数字信源
        /// </summary>
        /// <param name="packet"></param>
        private void DataCacheJson(PacketInfo packetInfo, IBatch batch, RedisHelper redis)
        {
            try
            {
                Dictionary<string, Object> data = packetInfo.structData;
                //判断最新时间是否大于redis保存的时间 大于则返回，否则会造成数据跳动
                //仅仅用于记录最新时间
                string redisKey = "newest_time_cache_" + packetInfo.positionId;
                DateTime newestSetTime = redis.StringGet<DateTime>(redisKey);
                if ((newestSetTime < packetInfo.receivingTime))
                {
                    redis.StringSet(redisKey, packetInfo.receivingTime);
                }

                //新增msgType用于判断
                redisKey = "newest_time_cache_" + packetInfo.positionId + "_" + packetInfo.header.msgType;
                newestSetTime = redis.StringGet<DateTime>(redisKey);
                if ((newestSetTime >= packetInfo.receivingTime))
                {
                    return;
                }
                redis.StringSet(redisKey, packetInfo.receivingTime);

                Dictionary<string, string> valueDict = new();
                //最新时间也写入
                valueDict.TryAdd("LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                valueDict.TryAdd("receiveNode", packetInfo.receiveNode);

                //前缀，避免告警和状态数据混淆
                //大写report、alarm
                //prefix为REPORT时，是状态数据，且数据量巨大，是单独返回到前台，所以key单独设置，以免影响prefix为ALARM时告警判断时获取数据慢
                List<string> typeKeys = data.Keys.Where(k => "TYPE".Equals(k.ToUpper())).ToList();
                string prefix = typeKeys.Count > 0 ? data[typeKeys[0]].ToString().ToUpper() : "REPORT";


                if (data != null && data.Count > 0)
                {
                    //解析
                    //GetDataFromJsonData(valueDict, data);
                    if ("REPORT".Equals(prefix))
                    {
                        GetDataFromJsonReport(valueDict, data);
                    }
                    else
                    {
                        GetDataFromJsonAlarm(valueDict, data);
                    }
                }

                //删除以前的垃圾数据
                //redis.KeyDelete("DATA_" + packetInfo.positionId);

                foreach (string key in valueDict.Keys)
                {
                    //batch.MHashSetAsync("DATA_" + packetInfo.positionId + ("REPORT".Equals(prefix) ? "_" + prefix : ""), packetInfo.positionId + "_" + key.Replace("公共", "COMMON").ToUpper(), valueDict[key]);
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId + ("REPORT".Equals(prefix) ? "_" + prefix : ""), key.Replace("公共", "COMMON").ToUpper(), valueDict[key]);
                }

                //20240719 码流记录仪数据特殊处理
                if ("REPORT".Equals(prefix) && valueDict.ContainsKey("AV_ALARM") &&
                    valueDict["AV_ALARM"] != null && !"".Equals(valueDict["AV_ALARM"].ToString()))
                {
                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, "LATEST_TIME_AV_ALARM", TransForm.GetFormatDatetime());

                    string avAlarm = valueDict["AV_ALARM"].ToString();
                    Dictionary<string, Dictionary<string, Dictionary<string, Object>>> dataAlarm = JsonConvert.DeserializeObject<Dictionary<string, Dictionary<string, Dictionary<string, Object>>>>(avAlarm);
                    foreach (string pdId in dataAlarm.Keys)
                    {
                        //只要开路接收频点1、2的数据
                        if ("6".Equals(pdId) || "7".Equals(pdId))
                        {
                            //节目编号
                            foreach (string pNo in dataAlarm[pdId].Keys)
                            {
                                string keyPrefix = pNo + "_" + pdId + "_AV_ALARM";
                                string programName = dataAlarm[pdId][pNo]["program_name"].ToString();
                                string pid = dataAlarm[pdId][pNo]["pid"].ToString();

                                batch.MHashSetAsync("DATA_" + packetInfo.positionId, "PROGRAM_" + keyPrefix, programName);
                                batch.MHashSetAsync("DATA_" + packetInfo.positionId, "SID_" + keyPrefix, pid);

                                Dictionary<string, Dictionary<string, string>> alarmInfo = JsonConvert.DeserializeObject<Dictionary<string, Dictionary<string, string>>>(dataAlarm[pdId][pNo]["alarm_info"].ToString());
                                if (alarmInfo != null)
                                {
                                    foreach (string alarmId in alarmInfo.Keys)
                                    {
                                        string val = "START".Equals(alarmInfo[alarmId]["startOrEnd"].ToUpper()) ? "1" : "0";
                                        batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, val);
                                    }

                                    foreach (string alarmId in avAlarmCodes.Keys)
                                    {
                                        if (!alarmInfo.ContainsKey(alarmId))
                                        {
                                            batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, "0");
                                        }
                                    }
                                }
                                else
                                {
                                    //告警编号没有的，恢复不告警
                                    foreach (string alarmId in avAlarmCodes.Keys)
                                    {
                                        batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, "0");
                                    }
                                }
                            }

                            //节目编号没有的，恢复不告警
                            foreach (string pNo in avPdIds[pdId].Keys)
                            {
                                if (!dataAlarm[pdId].ContainsKey(pNo))
                                {
                                    string keyPrefix = pNo + "_" + pdId + "_AV_ALARM";
                                    foreach (string alarmId in avAlarmCodes.Keys)
                                    {
                                        batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, "0");
                                    }
                                }
                            }
                        }
                    }

                    //整个频点都没有的，全部恢复
                    foreach (string pdId in avPdIds.Keys)
                    {
                        if (!dataAlarm.ContainsKey(pdId))
                        {
                            foreach (string pNo in avPdIds[pdId].Keys)
                            {
                                string keyPrefix = pNo + "_" + pdId + "_AV_ALARM";
                                foreach (string alarmId in avAlarmCodes.Keys)
                                {
                                    batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, "0");
                                }
                            }
                        }
                    }
                }

                //没有AV_ALARM返回的，全部恢复
                if ("REPORT".Equals(prefix) && (!valueDict.ContainsKey("AV_ALARM") ||
                    valueDict["AV_ALARM"] == null || "".Equals(valueDict["AV_ALARM"].ToString())))
                {
                    //整个频点都没有的，全部恢复
                    foreach (string pdId in avPdIds.Keys)
                    {
                        foreach (string pNo in avPdIds[pdId].Keys)
                        {
                            string keyPrefix = pNo + "_" + pdId + "_AV_ALARM";
                            foreach (string alarmId in avAlarmCodes.Keys)
                            {
                                batch.MHashSetAsync("DATA_" + packetInfo.positionId, alarmId + "_" + keyPrefix, "0");
                            }
                        }
                    }
                }

                batch.MExecute();
            }
            catch (Exception e)
            {
                mLogger.LogError("json写redis缓存数据失败！ " + e);
            }
        }

        /// <summary>
        /// xml
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <param name="batch"></param>
        /// <param name="redis"></param>
        private void DataCacheXml(PacketInfo packetInfo, IBatch batch, RedisHelper redis)
        {
            try
            {
                ////判断最新时间是否大于redis保存的时间 大于则返回，否则会造成数据跳动
                ////仅仅用于记录最新时间
                //string redisKey = "newest_time_cache_" + packetInfo.positionId;
                //DateTime newestSetTime = redis.StringGet<DateTime>(redisKey);
                //if ((newestSetTime < packetInfo.receivingTime))
                //{
                //    redis.StringSet(redisKey, packetInfo.receivingTime);
                //}

                ////新增msgType用于判断
                //redisKey = "newest_time_cache_" + packetInfo.positionId + "_" + packetInfo.header.msgType;
                //newestSetTime = redis.StringGet<DateTime>(redisKey);
                //if ((newestSetTime >= packetInfo.receivingTime))
                //{
                //    return;
                //}
                //redis.StringSet(redisKey, packetInfo.receivingTime);

                ////最新时间也写入
                //batch.MHashSetAsync("DATA_" + packetInfo.positionId, "LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                //batch.MHashSetAsync("DATA_" + packetInfo.positionId, "receiveNode", packetInfo.receiveNode);
                //if (packetInfo.dictData != null && packetInfo.dictData.Count > 0)
                //{
                //    Dictionary<string, string> dictDatas = packetInfo.dictData;
                //    foreach (string key in dictDatas.Keys)
                //    {
                //        if (dictDatas[key] != null && !"".Equals(dictDatas[key]) && !"NULL".Equals(dictDatas[key]))
                //        {
                //            batch.MHashSetAsync("DATA_" + packetInfo.positionId, key.ToUpper(), dictDatas[key] ?? "");
                //        }
                //    }
                //}

                List<string> positionIds = packetInfo.structData != null ? packetInfo.structData.Keys.ToList() : new List<string>() { packetInfo.positionId };
                foreach (string positionId in positionIds)
                {
                    //判断最新时间是否大于redis保存的时间 大于则返回，否则会造成数据跳动
                    //仅仅用于记录最新时间
                    string redisKey = "newest_time_cache_" + positionId;
                    DateTime newestSetTime = redis.StringGet<DateTime>(redisKey);
                    if ((newestSetTime < packetInfo.receivingTime))
                    {
                        redis.StringSet(redisKey, packetInfo.receivingTime);
                    }

                    //新增msgType用于判断
                    redisKey = "newest_time_cache_" + positionId + "_" + packetInfo.header.msgType;
                    newestSetTime = redis.StringGet<DateTime>(redisKey);
                    if ((newestSetTime >= packetInfo.receivingTime))
                    {
                        return;
                    }
                    redis.StringSet(redisKey, packetInfo.receivingTime);

                    //最新时间也写入
                    batch.MHashSetAsync("DATA_" + positionId, "LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                    batch.MHashSetAsync("DATA_" + positionId, "DATA_FROM", packetInfo.ip);
                    batch.MHashSetAsync("DATA_" + positionId, "receiveNode", packetInfo.receiveNode);

                    //把最新的台站远场IP记录下来
                    string stationId = positionId.Substring(0, positionId.IndexOf("_"));
                    //用memo存储设备IP
                    string dev_ip = packetInfo.header.memo;
                    string tmpStr = (("".Equals(dev_ip) || !dev_ip.Contains(".") || !TransForm.IsIP(dev_ip)) ? "" : "#" + dev_ip);
                    batch.MHashSetAsync("stationId2YcIpHash", stationId, packetInfo.ip + tmpStr);
                }

                if (packetInfo.dictData != null && packetInfo.dictData.Count > 0)
                {
                    Dictionary<string, string> dictDatas = packetInfo.dictData;
                    foreach (string key in dictDatas.Keys)
                    {
                        //key已经提拼接上了前缀，以#分隔
                        if (dictDatas[key] != null && !"".Equals(dictDatas[key]) && !"NULL".Equals(dictDatas[key]))
                        {
                            string rootKey = key.Split("#")[0];
                            string dataKey = key.Split("#").Length > 1 ? key.Split("#")[1] : key;
                            batch.MHashSetAsync(rootKey, dataKey, dictDatas[key] ?? "");
                        }
                    }
                }

                batch.MExecute();
            }
            catch (Exception e)
            {
                mLogger.LogError("xml写redis缓存数据失败！ " + e);
            }
        }

        /// <summary>
        /// 遍历json数据
        /// 简洁处理，不用GetDataFromJsonData
        /// 状态数据，只处理到首层节点
        /// </summary>
        /// <param name="valueDict"></param>
        /// <param name="data"></param>
        private void GetDataFromJsonReport(Dictionary<string, string> valueDict, Object data)
        {
            try
            {
                if (data is Dictionary<string, Object>)
                {
                    Dictionary<string, Object> dataDict = (Dictionary<string, Object>)data;
                    foreach (string key in dataDict.Keys)
                    {
                        string keyUpper = key.ToUpper();
                        if (!valueDict.ContainsKey(keyUpper))
                        {
                            valueDict.TryAdd(keyUpper, JsonConvert.SerializeObject(dataDict[key]));
                        }
                    }
                }
            }
            catch { }
        }

        /// <summary>
        /// 遍历json数据
        /// 告警数据，list层级数据直接当初json返回，其它层级一层一层拼接处理
        /// </summary>
        /// <param name="valueDict"></param>
        /// <param name="data"></param>
        /// <param name="prefix"></param>
        private void GetDataFromJsonAlarm(Dictionary<string, string> valueDict, Object data, string prefix = "")
        {
            try
            {
                if (basicType.ContainsKey(data.GetType().Name))
                {
                    if (!"".Equals(data.ToString()) && !"NULL".Equals(data.ToString()))
                    {
                        valueDict.TryAdd("".Equals(prefix) ? data.GetType().Name : prefix, data.ToString());
                    }
                }
                else if (data is Dictionary<string, Object>)
                {
                    Dictionary<string, Object> dataDict = (Dictionary<string, Object>)data;
                    foreach (string key in dataDict.Keys)
                    {
                        Object dataIn = dataDict[key];
                        string name = key + ("".Equals(prefix) ? "" : "_" + prefix);
                        if (dataIn != null)
                        {
                            GetDataFromJsonAlarm(valueDict, dataIn, name);
                        }
                    }
                }
                else if (data is List<Object>)
                {
                    //List<Object> dataList = (List<Object>)data;
                    //int i = 0;
                    //foreach (var dataIn in dataList)
                    //{
                    //    string name = i + ("".Equals(prefix) ? "" : "_" + prefix);
                    //    if (dataIn != null)
                    //    {
                    //        GetDataFromJsonAlarm(valueDict, dataIn, name);
                    //    }
                    //    i++;
                    //}

                    //告警明细list，直接返回对象json字符串
                    valueDict.TryAdd("".Equals(prefix) ? data.GetType().Name : prefix, JsonConvert.SerializeObject(data));
                }
                else if (data is JObject)
                {
                    Dictionary<string, Object> dataChild = JsonConvert.DeserializeObject<Dictionary<string, Object>>(data.ToString());
                    if (dataChild != null)
                    {
                        GetDataFromJsonAlarm(valueDict, dataChild, prefix);
                    }
                }
                else if (data is JArray)
                {
                    List<Object> dataChild = JsonConvert.DeserializeObject<List<Object>>(data.ToString());
                    if (dataChild != null)
                    {
                        GetDataFromJsonAlarm(valueDict, dataChild, prefix);
                    }
                }
                else
                {
                    if (!"".Equals(data.ToString()) && !"NULL".Equals(data.ToString()))
                    {
                        valueDict.TryAdd("".Equals(prefix) ? data.GetType().Name : prefix, data.ToString());
                    }
                }
            }
            catch { }
        }

        /// <summary>
        /// 通用json遍历
        /// </summary>
        /// <param name="valueDict"></param>
        /// <param name="data"></param>
        /// <param name="prefix"></param>
        /// <param name="gap"></param>
        private Dictionary<string, string> GetDataFromJson(Dictionary<string, string> valueDict, Object data, string prefix = "", string gap = "#")
        {
            try
            {
                if (basicType.ContainsKey(data.GetType().Name))
                {
                    if (!"".Equals(data.ToString()) && !"NULL".Equals(data.ToString()))
                    {
                        valueDict.TryAdd(("".Equals(prefix) ? data.GetType().Name : prefix).ToUpper(), data.ToString());
                    }
                }
                else if (data is Dictionary<string, Object>)
                {
                    Dictionary<string, Object> dataDict = (Dictionary<string, Object>)data;
                    foreach (string key in dataDict.Keys)
                    {
                        Object dataIn = dataDict[key];
                        string name = ("".Equals(prefix) ? "" : prefix + gap) + key;
                        if (dataIn != null)
                        {
                            GetDataFromJson(valueDict, dataIn, name);
                        }
                    }
                }
                else if (data is List<Object>)
                {
                    List<Object> dataList = (List<Object>)data;
                    int i = 0;
                    foreach (var dataIn in dataList)
                    {
                        string name = ("".Equals(prefix) ? "" : prefix + gap) + i;
                        if (dataIn != null)
                        {
                            GetDataFromJson(valueDict, dataIn, name);
                        }
                        i++;
                    }
                }
                else if (data is JObject)
                {
                    Dictionary<string, Object> dataChild = JsonConvert.DeserializeObject<Dictionary<string, Object>>(data.ToString());
                    if (dataChild != null)
                    {
                        GetDataFromJson(valueDict, dataChild, prefix);
                    }
                }
                else if (data is JArray)
                {
                    List<Object> dataChild = JsonConvert.DeserializeObject<List<Object>>(data.ToString());
                    if (dataChild != null)
                    {
                        GetDataFromJson(valueDict, dataChild, prefix);
                    }
                }
                else
                {
                    if (!"".Equals(data.ToString()) && !"NULL".Equals(data.ToString()))
                    {
                        valueDict.TryAdd(("".Equals(prefix) ? data.GetType().Name : prefix).ToUpper(), data.ToString());
                    }
                }
            }
            catch { }

            return valueDict;
        }
        #endregion

        #region 辅助方法
        /// <summary>
        /// 获取设备信息
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <param name="seviceSN"></param>
        /// <returns></returns>
        private DeviceInfo GetDeviceInfo(PacketInfo packetInfo, string seviceSN = "")
        {
            //判断并记录设备信息
            //得到设备信息，是为了获取seviceSN设备序列号
            DeviceInfo deviceInfo;

            try
            {
                //先从缓存判断
                if (deviceInfos.ContainsKey(packetInfo.positionId))
                {
                    deviceInfo = deviceInfos[packetInfo.positionId];
                    deviceInfo.new_time = DateTime.Now;

                    //姑且认为端口、台站级别和台站类型有可能会变化吧，更新一下到redis
                    //position_id_node 本工位号从哪个采集节点接收数据的，采集服务重启时会改变
                    //if (!deviceInfo.port.ToString().Equals(packetInfo.port.ToString()) ||
                    if (!deviceInfo.staionLevel.Equals(packetInfo.stationLevel) ||
                        !deviceInfo.position_id_node.Equals(packetInfo.receiveNode) ||
                        (seviceSN != null && !"".Equals(seviceSN) && !deviceInfo.seviceSN.Equals(seviceSN)) ||
                        deviceInfo.data_type != 1
                        )
                    {
                        //deviceInfo.port = packetInfo.port;
                        deviceInfo.staionLevel = packetInfo.stationLevel;
                        deviceInfo.position_id_node = packetInfo.receiveNode;
                        deviceInfo.seviceSN = (seviceSN != null && !"".Equals(seviceSN) && !deviceInfo.seviceSN.Equals(seviceSN)) ? seviceSN : deviceInfo.seviceSN;
                        //转换成1
                        deviceInfo.data_type = 1;
                        redisGlobal.HashSet("station_deviceInfos", packetInfo.positionId, deviceInfo);
                    }
                }
                else
                {
                    //再从redis获取
                    deviceInfo = redisGlobal.HashGet<DeviceInfo>("station_deviceInfos", packetInfo.positionId);
                    if (deviceInfo == null)
                    {
                        //redis也没有，则新增要给，然后写入redis和记录到缓存
                        deviceInfo = new DeviceInfo
                        {
                            seviceSN = (seviceSN == null || "".Equals(seviceSN)) ? System.Guid.NewGuid().ToString("N") : seviceSN,
                            stationCode = packetInfo.stationCode,
                            staionLevel = packetInfo.stationLevel,
                            position_id = packetInfo.positionId,
                            position_id_node = packetInfo.receiveNode,
                            deviceType = packetInfo.header.device,
                            ip_addr = packetInfo.ip,
                            ip_type = TransForm.IsIPV6(packetInfo.ip) ? "1" : "0",
                            port = packetInfo.port,
                            dev_addr = packetInfo.header.devaddr.ToString(),
                            data_type = 1
                        };
                        deviceInfo.new_time = DateTime.Now;

                        redisGlobal.HashSet("station_deviceInfos", packetInfo.positionId, deviceInfo);
                    }
                    else
                    {
                        deviceInfo.new_time = DateTime.Now;
                    }

                    lock (syncLock)
                    {
                        //从redis获取后，放入缓存
                        deviceInfos.TryAdd(packetInfo.positionId, deviceInfo);
                    }
                }
            }
            catch
            {
                deviceInfo = null;
            }

            return deviceInfo;
        }

        /// <summary>
        /// 更新设备信息
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <param name="seviceSN"></param>
        private void CheckDeviceInfos(PacketInfo packetInfo, string seviceSN = "")
        {
            try
            {
                //判断并记录设备信息
                //得到设备信息，是为了获取seviceSN设备序列号
                DeviceInfo deviceInfo;

                //新远场的xml文件，可能存在多个台站公用一个设备的情况
                List<string> positionIds = (packetInfo.structData != null && "3".Equals(packetInfo.dataType)) ? packetInfo.structData.Keys.ToList() : new List<string>() { packetInfo.positionId };
                foreach (string positionId in positionIds)
                {
                    //先从缓存判断
                    if (deviceInfos.ContainsKey(positionId))
                    {
                        deviceInfo = deviceInfos[positionId];
                        deviceInfo.new_time = DateTime.Now;

                        //姑且认为端口、台站级别和台站类型有可能会变化吧，更新一下到redis
                        //position_id_node 本工位号从哪个采集节点接收数据的，采集服务重启时会改变
                        //if (!deviceInfo.port.ToString().Equals(packetInfo.port.ToString()) ||
                        if (!deviceInfo.staionLevel.Equals(packetInfo.stationLevel) ||
                            !deviceInfo.position_id_node.Equals(packetInfo.receiveNode) ||
                            (seviceSN != null && !"".Equals(seviceSN) && !deviceInfo.seviceSN.Equals(seviceSN)) ||
                            deviceInfo.data_type != 2
                            )
                        {
                            //deviceInfo.port = packetInfo.port;
                            deviceInfo.staionLevel = packetInfo.stationLevel;
                            deviceInfo.position_id_node = packetInfo.receiveNode;
                            deviceInfo.seviceSN = (seviceSN != null && !"".Equals(seviceSN) && !deviceInfo.seviceSN.Equals(seviceSN)) ? seviceSN : deviceInfo.seviceSN;
                            //转换成2
                            deviceInfo.data_type = 2;
                            redisGlobal.HashSet("station_deviceInfos", positionId, deviceInfo);
                        }
                    }
                    else
                    {
                        //再从redis获取
                        deviceInfo = redisGlobal.HashGet<DeviceInfo>("station_deviceInfos", positionId);
                        if (deviceInfo == null)
                        {
                            //redis也没有，则新增要给，然后写入redis和记录到缓存
                            deviceInfo = new DeviceInfo
                            {
                                seviceSN = (seviceSN == null || "".Equals(seviceSN)) ? System.Guid.NewGuid().ToString("N") : seviceSN,
                                stationCode = packetInfo.stationCode,
                                staionLevel = packetInfo.stationLevel,
                                position_id = positionId,
                                position_id_node = packetInfo.receiveNode,
                                deviceType = packetInfo.header.device,
                                ip_addr = packetInfo.ip,
                                ip_type = TransForm.IsIPV6(packetInfo.ip) ? "1" : "0",
                                port = packetInfo.port,
                                dev_addr = packetInfo.header.devaddr.ToString(),
                                data_type = 2
                            };
                            deviceInfo.new_time = DateTime.Now;

                            redisGlobal.HashSet("station_deviceInfos", positionId, deviceInfo);
                        }
                        else
                        {
                            deviceInfo.new_time = DateTime.Now;
                        }

                        lock (syncLock)
                        {
                            //从redis获取后，放入缓存
                            deviceInfos.TryAdd(positionId, deviceInfo);
                        }
                    }
                }
            }
            catch
            {
            }
        }
        #endregion

        #region 从接收缓存读取并处理
        /// <summary>
        /// 接收数据处理线程
        /// </summary>
        /// <param name="rcvDataCache"></param>
        /// <param name="tp"></param>
        private void DeaRcvlData(BlockingCollection<ReceiveData> rcvDataCache)
        {
            while (true)
            {
                try
                {
                    if (rcvDataCache.Count > 0)
                    {
                        while (rcvDataCache.Count > 0)
                        {
                            ReceiveData receiveData;
                            lock (rcvDataCache)
                            {
                                receiveData = rcvDataCache.Take();
                            }

                            //启动线程池
                            ThreadPool.QueueUserWorkItem(new WaitCallback(DealRcvDataPool), receiveData);
                        }
                    }
                }
                catch { }
                finally
                {
                    Thread.Sleep(1);
                }
            }
        }

        /// <summary>
        /// 处理数据
        /// </summary>
        /// <param name="o"></param>
        private void DealRcvDataPool(Object o)
        {
            try
            {
                ReceiveData receiveData = (ReceiveData)o;
                Dictionary<string, Object> rs = (Dictionary<string, Object>)receiveData.func?.Invoke(receiveData);
                if (int.Parse(rs["return_code"].ToString()) != 0)
                {
                    //string msg = receiveData.type == 0 ? "数字信源" : ("科江新远场：" + receiveData.data);
                    string msg = "";
                    if (receiveData.type == 0)
                    {
                        msg = "数字信源";
                    }
                    else if (receiveData.type == 1)
                    {
                        msg = "科江新远场：" + receiveData.data;
                    }
                    else if (receiveData.type == 2 || receiveData.type == 3)
                    {
                        msg = "监控系统WEBAPI接口协议";
                    }
                    string from = receiveData.ip + ":" + receiveData.port;
                    mLogger.LogError("处理来自于【" + from + "】的【" + msg + "】数据异常，" + rs["return_msg"]);
                }
            }
            catch { }
        }
        #endregion

        #region 定时从总平台获取数据
        /// <summary>
        /// 定时从总平台获取数据
        /// </summary>
        private void GetFromSvr()
        {
            mLogger.LogInformation("定时从总平台获取数据线程启动");
            Thread.Sleep(1000);

            //定时获取间隔
            int interval = 2000;
            Stopwatch sw;
            while (true)
            {
                sw = Stopwatch.StartNew();
                try
                {
                    int count = 0;
                    //新远场数据
                    string ycUrl = webDataUrl + (webDataUrl.EndsWith("/") ? "" : "/") + "getyc?stationId=" + stationIdGlobal;
                    Dictionary<string, Object> ycResult = httpClientUtil.Get<Dictionary<string, Object>>(ycUrl);
                    if (ycResult != null && int.Parse(ycResult["return_code"].ToString() ?? "") == 0 && ycResult.ContainsKey("data") && !string.IsNullOrEmpty(ycResult["data"].ToString()) && ValidJson.IsJson(ycResult["data"].ToString()))
                    {
                        Dictionary<string, string> ycData = JsonConvert.DeserializeObject<Dictionary<string, string>>(ycResult["data"].ToString());
                        if (ycData != null)
                        { 
                            foreach (string dataKey in ycData.Keys)
                            {
                                svrBatch.MHashSetAsync("DATA_" + stationIdGlobal + "_0x0902_3", dataKey, ycData[dataKey]);
                                count++;
                            }
                        }
                    }

                    //地数图像识别数据
                    string aiUrl = webDataUrl + (webDataUrl.EndsWith("/") ? "" : "/") + "getai?stationId=" + stationIdGlobal;
                    Dictionary<string, Object> aiResult = httpClientUtil.Get<Dictionary<string, Object>>(aiUrl);
                    if (aiResult != null && int.Parse(aiResult["return_code"].ToString() ?? "") == 0 && aiResult.ContainsKey("data") && !string.IsNullOrEmpty(aiResult["data"].ToString()) && ValidJson.IsJson(aiResult["data"].ToString()))
                    {
                        Dictionary<string, string> aiData = JsonConvert.DeserializeObject<Dictionary<string, string>>(aiResult["data"].ToString());
                        if (aiData != null)
                        { 
                            foreach (string dataKey in aiData.Keys)
                            {
                                svrBatch.MHashSetAsync("DATA_" + stationIdGlobal + "_0x1003_1", dataKey, aiData[dataKey]);
                                count++;
                            }
                        }
                    }

                    //中波数据
                    string zbUrl = webDataUrl + (webDataUrl.EndsWith("/") ? "" : "/") + "getzb?stationId=" + stationIdGlobal;
                    Dictionary<string, Object> zbResult = httpClientUtil.Get<Dictionary<string, Object>>(zbUrl);
                    if (zbResult != null && int.Parse(zbResult["return_code"].ToString() ?? "") == 0 && zbResult.ContainsKey("data") && !string.IsNullOrEmpty(zbResult["data"].ToString()) && ValidJson.IsJson(zbResult["data"].ToString()))
                    {
                        Dictionary<string, string> zbData = JsonConvert.DeserializeObject<Dictionary<string, string>>(zbResult["data"].ToString());
                        if (zbData != null)
                        {
                            foreach (string dataKey in zbData.Keys)
                            {
                                svrBatch.MHashSetAsync("DATA_" + stationIdGlobal + "_0x0903_1", dataKey, zbData[dataKey]);
                                count++;
                            }
                        }
                    }

                    if (count > 0)
                    {
                        svrBatch.Execute();
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogInformation("从总平台获取数据异常，" + e.Message);
                }
                finally
                {
                    int gap = (int)sw.ElapsedMilliseconds;
                    Thread.Sleep(gap >= interval ? 1 : (interval - gap));
                }
            }
        }
        #endregion

        #region 写redis，写台站、设备在线信息
        /// <summary>
        /// 更新台站和设备在线信息数据
        /// </summary>
        private void UpdateOnlineStatus()
        {
            while (true)
            {
                try
                {
                    while (targetCache.Count > 0)
                    {
                        string pkgInfo;
                        IBatch batch = redisGlobal.CreateBatch();
                        int cnt = 0;
                        while (targetCache.Count > 0)
                        {
                            lock (targetCache)
                            {
                                pkgInfo = targetCache.Take();
                            }

                            //太多来不及了，先丢弃
                            if (targetCache.Count > 3000)
                            {
                                continue;
                            }
                            string ip = pkgInfo.Split(",")[0];
                            string positionId = pkgInfo.Split(",")[1];
                            string rcvTime = pkgInfo.Split(",")[2].Replace("\"", "");
                            //台站数据是所有数据，适当丢弃一些也是可以的，所以有条件的丢弃
                            if (targetCache.Count < 1000)
                            {
                                if (targetStation.ContainsKey(ip))
                                {
                                    batch.MStringSetAsync("online_" + targetStation[ip], "0");
                                    batch.MStringSetAsync("newest_time_" + targetStation[ip], rcvTime);
                                }
                            }

                            //工位数据
                            if (!"".Equals(positionId))
                            {
                                batch.MStringSetAsync("online_" + positionId, "0");
                                batch.MStringSetAsync("newest_time_" + positionId, rcvTime);
                            }
                            cnt++;

                            if (cnt > 50)
                            {
                                batch.MExecute();
                                batch = redisGlobal.CreateBatch();
                                cnt = 0;
                            }
                        }
                        if (cnt > 0)
                        {
                            batch.MExecute();
                        }
                    }
                }
                catch { }
                finally
                {
                    Thread.Sleep(1);
                }
            }
        }
        #endregion

        #region 心跳服务
        /// <summary>
        /// 心跳服务，定时往redis发送心跳数据
        /// 用于获取当前服务副本数量
        /// </summary>
        private void HeartBeatTask()
        {
            string lockKey = "lockKey" + keySvr;   //心跳节点锁key，保证唯一
            string uuid = nodeName;
            int gapSec = 2; //循环判断间隔
            int gapCheck = 20;   //判断时间差间隔，适当大于gapSec，如果大于，则判断为该节点离线，移除掉

            //每次启动清空
            Thread.Sleep(new Random().Next(1, (gapSec + 1)) * 1000);

            //开始先写一条服务启动时间，用于给离线判断服务判断启动时间2分钟内不做离线判断，避免重启采集服务造成台站上报超时的告警
            //TimeSpan.FromMinutes(2) key存在分钟数
            redisInit.StringSet("DataReceiveService_start", "", TimeSpan.FromMinutes(2));
            while (true)
            {
                try
                {
                    DateTime dtNow = DateTime.Now;

                    //更新自身节点最新时间
                    redisInit.HashSet(keySvr, uuid, dtNow);

                    //记录统计信息，也作为心跳数据记录的一部分
                    nodeStatusInfo["Datetime"] = TransForm.DateTimeFormat(dtNow);
                    nodeStatusInfo["GetSum"] = numsRcv;
                    nodeStatusInfo["SkipNum"] = skipNum;
                    nodeStatusInfo["nodeNo"] = nodeNo + 1;
                    redisInit.HashSet("STATUS_" + keySvr, uuid, nodeStatusInfo);

                    //判断副本数最新时间 锁过期时间为5秒
                    redisInit.LockAndDo(lockKey, uuid, TimeSpan.FromSeconds(5),
                        () =>
                        {
                            //获取key下所有节点及其时间
                            Dictionary<string, DateTime> heartInfo = redisInit.HashGetAllDict<DateTime>(keySvr);
                            if (heartInfo != null && heartInfo.Count > 0)
                            {
                                //判断节点是否超时gapCheck，超时则移除
                                //判断当前节点ID再心跳节点的位置
                                int idx = 0;
                                foreach (string node in heartInfo.Keys)
                                {
                                    if (uuid.Equals(node))
                                    {
                                        //节点ID位置
                                        nodeNo = idx;
                                    }

                                    //判断节点过期时间
                                    if ((DateTime.Now - heartInfo[node]).TotalSeconds > gapCheck)
                                    {
                                        redisInit.HashDelete(keySvr, node);
                                        redisInit.HashDelete("STATUS_" + keySvr, node);
                                    }
                                    else
                                    {
                                        idx++;
                                    }
                                }

                                //获得节点数
                                nodeNum = idx;
                            }
                            return true;
                        });
                }
                finally
                {
                    Thread.Sleep(gapSec * 1000);
                }
            }
        }
        #endregion

        #region 日志打印
        /// <summary>
        /// 定时监视判断是否增加线程池数量来处理数据
        /// </summary>
        private void TimerTask()
        {
            int times = 0;
            long rate;    //数量差
            long numsRcvPre = 0;    //上次数量

            while (true)
            {
                rate = numsRcv - numsRcvPre;
                numsRcvPre = numsRcv;

                //以十秒钟为单位，统计接收频率
                nodeStatusInfo["GetRate"] = rate;

                //数据有变化，打印日志
                int redisCount = dataCacheToRedis.Count;
                if (rate > 0 || gapNums > 0 || skipNum > 0 || redisCount > 0 || WebsocketClientManager.websocketClient.GetCount() > 0)
                {
                    mLogger.LogInformation(nodeName + 
                        " 线程数=" + ThreadPool.ThreadCount +
                        " WS连接=" + WebsocketClientManager.websocketClient.GetCount() +
                        " 数据帧=" + numsRcv.ToString() +
                        " 接收帧率=" + rate + " 缓存帧=" + gapNums + " 丢弃=" + skipNum +
                        " 写redis缓存=" + dataCacheToRedis.Count +
                        " 节点数=" + nodeNum + " 当前节点号=" + (nodeNo + 1));
                }
                else
                {
                    mLogger.LogInformation(nodeName +
                        " 线程数=" + ThreadPool.ThreadCount +
                        " WS连接=" + WebsocketClientManager.websocketClient.GetCount() +
                        " 节点数=" + nodeNum + " 当前节点号=" + (nodeNo + 1));
                }

                lock (skipLock)
                {
                    skipNum = 0;
                }
                if (exRcv != null)
                {
                    mLogger.LogError("接收数据有异常，" + exRcv);
                    lock (exLock)
                    {
                        exRcv = null;
                    }
                }

                //十分钟更新一次缓存
                if (times % 6 == 0)
                {
                    InitCacheInfo();
                }

                times++;
                if (times > 10 * 10 * 10 * 10000)
                {
                    times = 1;
                }

                try
                {
                    ycBatch.MExecute();
                }
                catch { }

                Thread.Sleep(10 * 1000);
            }
        }
        #endregion

        #region 配置数据主动采集
        /// <summary>
        /// 轮询设备配置项，并进行比对
        /// </summary>
        private void PollingSetting()
        {
            mLogger.LogInformation("配置读取线程启动");
            Stopwatch sw;
            //判断间隔
            int interval = 1000;
            Thread.Sleep(interval);
            while (true)
            {
                sw = Stopwatch.StartNew();
                try
                {
                    //新主控，只需要主动抓配置数据就行
                    Dictionary<string, DateTime> stationSpeed = redisGlobal.HashGetAllDict<DateTime>("stationSpeedCfg");
                    if (stationSpeed.ContainsKey(stationIdGlobal) && stationPositionDeviceInfos.ContainsKey(stationIdGlobal) && stationPositionDeviceInfos[stationIdGlobal].Count > 0)
                    {
                        Dictionary<string, DeviceInfo> stationDevices = stationPositionDeviceInfos[stationIdGlobal];
                        foreach (string key in cmdSetInfo.Keys)
                        {
                            DateTime dtNow = DateTime.Now;

                            string setId = cmdSetInfo[key]["set_id"];
                            string devType = cmdSetInfo[key]["device_type"];
                            string positionId = cmdSetInfo[key]["position_id"];
                            string getType = cmdSetInfo[key]["get_type"];
                            string setName = cmdSetInfo[key]["set_name"];

#if DEBUG
                            bool positionOffLine = false;
#else
                            //数据来自旧主控时，不进行抓取
                            if (GetPositionFromUdp(positionId))
                            {
                                continue;
                            }

                            //先从缓存判断
                            if (!deviceInfos.ContainsKey(positionId))
                            {
                                continue;
                            }

                            bool positionOffLine = GetPositionOffLine(positionId);
#endif

                            if (!positionOffLine && stationDevices.ContainsKey(positionId))
                            {
                                //构建请求
                                WebApiDataReqWS ContentReq = new WebApiDataReqWS()
                                {
                                    Command = "getdata",
                                    Data = new Dictionary<string, string>()
                                    {
                                        {"position_id",positionId },
                                        {"operate_type",getType }
                                    },
                                    Type = "data"
                                };
                                WebApiDataFull webApiDataFull = new WebApiDataFull(true) { webParamData = new WebParamData() { positionid = positionId, nowmal = true }, CheckParam = false, ContentReq = ContentReq };
                                bool rs = GetDataFromDev(webApiDataFull);
                                if (rs)
                                {
                                    webApiDataFull.dictData = webApiDataFull.ContentRes.SelfToDictionary();
                                    ReceiveData receiveData = new ReceiveData()
                                    {
                                        func = GetDataJson,
                                        data = webApiDataFull,
#if DEBUG
                                        ip = "127.0.0.1",
                                        port = 0,
#else
                                        ip = deviceInfos[positionId].ip_addr,
                                        port = (ushort)deviceInfos[positionId].port,
#endif
                                        type = 2
                                    };

                                    lock (jsonCache)
                                    {
                                        jsonCache.Add(receiveData);
                                    }
                                }
                                else
                                {
                                    mLogger.LogError($"从{positionId}获取{setName}失败，原因：{((WebApiDataRes)webApiDataFull.ContentRes).Message}");
                                }
                            }
                        }
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogError("轮询异常，" + e);
                }
                finally
                {
                    int gap = (int)sw.ElapsedMilliseconds + 10;
                    Thread.Sleep(gap >= interval ? 1 : (interval - gap));
                }
            }
        }

        /// <summary>
        /// 从设备获取数据
        /// </summary>
        /// <param name="webApiDataFull"></param>
        /// <returns></returns>
        private bool GetDataFromDev(WebApiDataFull webApiDataFull)
        {
            bool rs = false;

            try
            {
                //webApiDataFull.positionid应该要和要发往的目标positionid一样，否则无法匹配
                if (webApiDataFull.ContentReq.Data != null && (webApiDataFull.ContentReq.Data is JObject || webApiDataFull.ContentReq.Data is Dictionary<string, string> || webApiDataFull.ContentReq.Data is string))
                {
                    Dictionary<string, string> dataDetail;
                    if (webApiDataFull.ContentReq.Data is JObject)
                    {
                        dataDetail = JsonConvert.DeserializeObject<Dictionary<string, string>>(webApiDataFull.ContentReq.Data.ToString()).ToDictionary(d => d.Key.ToUpper(), d => d.Value);
                    }
                    else if (webApiDataFull.ContentReq.Data is Dictionary<string, string>)
                    {
                        dataDetail = (((Dictionary<string, string>)webApiDataFull.ContentReq.Data).ToDictionary(d => d.Key.ToUpper(), d => d.Value));
                    }
                    else
                    {
                        dataDetail = JsonConvert.DeserializeObject<Dictionary<string, string>>(webApiDataFull.ContentReq.Data.ToString()).ToDictionary(d => d.Key.ToUpper(), d => d.Value);
                    }

                    //必须携带positionid参数，否则无法定位到设备或主控
                    if (dataDetail != null && dataDetail.ContainsKey("POSITION_ID"))
                    {
                        string targetPositionId = dataDetail["POSITION_ID"].ToString();
                        WebsocketClient websocketClient = WebsocketClientManager.websocketClient.GetByPositionId(targetPositionId);
                        if (websocketClient != null)
                        {
                            if (websocketClient.webSocket.State == WebSocketState.Open)
                            {
                                string sendData = JsonConvert.SerializeObject(webApiDataFull.ContentReq);
                                byte[] responseBuffer = Encoding.UTF8.GetBytes(sendData);
                                websocketClient.webSocket.SendAsync(new ArraySegment<byte>(responseBuffer), System.Net.WebSockets.WebSocketMessageType.Text, true, CancellationToken.None);

                                //设置一个任务
                                TaskCompletionSource<Object> tcs = TcsTimeoutHelper.tcsManager.SetTcs(webApiDataFull.ContentReq.RequestId);
                                int tm = 3;
#if DEBUG
                                //方便测试
                                tm = 30;
#endif
                                //读取、操作都走WebSocket
                                webApiDataFull.ContentRes = (WebApiDataResWS)TcsTimeoutHelper.WaitWithTimeout(tcs, TimeSpan.FromSeconds(tm)).Result;
                                TcsTimeoutHelper.tcsManager.RemoveTcs(webApiDataFull.ContentReq.RequestId);
                                rs = true;
                            }
                            else
                            {
                                ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5501;
                                ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "，和设备" + targetPositionId + "连接已断开";
                            }
                        }
                        else
                        {
                            ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_404;
                            ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "，无法匹配到设备" + targetPositionId;
                        }
                    }
                    else
                    {
                        ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_400;
                        ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "，Data数据必须携带position_id参数";
                    }
                }
                else
                {
                    ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_400;
                    ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "，Data参数异常";
                }
            }
            catch (Exception e)
            {
                ((WebApiDataRes)webApiDataFull.ContentRes).Code = WebApiStatusCode.code_5503;
                ((WebApiDataRes)webApiDataFull.ContentRes).Message = EnumHelper.GetDescription(((WebApiDataRes)webApiDataFull.ContentRes).Code) + "," + e.Message;
            }

            return rs;
        }

        /// <summary>
        /// 获取工位号在线状态
        /// </summary>
        /// <param name="positionId"></param>
        /// <returns></returns>
        private bool GetPositionOffLine(string positionId)
        {
            //是否离线
            //string val = mRedisCache.GetString("online_" + positionId);
            string val = redisGlobal.StringGet("online_" + positionId);
            //获取不到值也判断为离线
            if (val != null && "0".Equals(val))
            {
                return false;
            }
            return true;
        }

        /// <summary>
        /// 获取工位号是否来自udp，即旧主控的UDP
        /// </summary>
        /// <param name="positionId"></param>
        /// <returns></returns>
        private bool GetPositionFromUdp(string positionId)
        {
            string jsonFlag = redisGlobal.HashGet("DATA_" + positionId, "json_flag");
            if (string.IsNullOrEmpty(jsonFlag) || "0".Equals(jsonFlag))
            {
                return true;
            }

            return false;
        }
#endregion

        /// <summary>
        /// 通用状态返回
        /// </summary>
        /// <returns></returns>
        public CommonDataRes GetStatus()
        {
            CommonDataRes commonDataRes = new CommonDataRes();

            Dictionary<string, Object> status = new();
            status.TryAdd("wsCount", WebsocketClientManager.websocketClient.GetCount());
            status.TryAdd("cacheCount", dataCacheToRedis.Count);
            status.TryAdd("waitCount", gapNums);
            commonDataRes.Data = status;

            return commonDataRes;
        }

        /// <summary>
        /// 获取客户端IP
        /// </summary>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public string GetClientIp(HttpContext httpContext)
        {
            string ip = "";
            try
            {
                int agentType = GetAgentType(httpContext);
                //web版本访问时，由于科江使用了代理，特定传输了这两个参数
                if (agentType == 2)
                {
                    ip = httpContext.Request.Headers["X-Real-IP"].ToString();
                    if (string.IsNullOrEmpty(ip))
                    {
                        httpContext.Request.Headers["X-Forwarded-For"].ToString();
                    }
                    if (string.IsNullOrEmpty(ip))
                    {
                        ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("F", "").Replace(":", "");
                    }
                }
                else
                {
                    ip = httpContext.Connection.RemoteIpAddress == null ? "" : httpContext.Connection.RemoteIpAddress.ToString().ToUpper().Replace("F", "").Replace(":", "");
                }
            }
            catch (Exception e)
            { }

            return ip;
        }

        /// <summary>
        /// 获取客户端类型
        /// 0-是测试的Apifox或者Postman
        /// 1-是cs客户端（即User-Agent为空串）
        /// 2-是bs客户端
        /// </summary>
        /// <param name="httpContext"></param>
        /// <returns></returns>
        public int GetAgentType(HttpContext httpContext)
        {
            int agentType = 0;
            try
            {
                string agent = httpContext.Request.Headers["User-Agent"].ToString();

                //是否cs客户端，用以区分是否cs客户端，因为web客户端的布局和cs客户端的不一样，需要区别开来
                //cs客户端过来的agent为空串
                //POSTMAN和APIFOX是测试
                //CSCLIENT是和客户端约定的字符串
                //agentType = (agent.ToUpper().Contains("POSTMAN") || agent.ToUpper().Contains("APIFOX")) ? 0 : ("CSCLIENT".Equals(agent.ToUpper()) ? 1 : 2);
                agentType = (agent.ToUpper().Contains("POSTMAN") || agent.ToUpper().Contains("APIFOX")) ? 0 : ("".Equals(agent.ToUpper()) ? 1 : 2);
            }
            catch (Exception e)
            { }

            return agentType;
        }

        /// <summary>
        /// 统计数据写入redis
        /// </summary>
        private void Statistics()
        {
            Stopwatch sw;
            //判断间隔
            int interval = 1000;
            long numsRcvPre = numsRcv;    //上次数量
            string keyPre = "";

            Thread.Sleep(interval);
            while (true)
            {
                sw = Stopwatch.StartNew();
                try
                {
                    long tmpNum = numsRcv;
                    long secondCount = tmpNum - numsRcvPre;
                    numsRcvPre = tmpNum;

                    DateTime dt = DateTime.Now;

                    //记录最近两年的数据
                    //以哈希存储，以天为key，小时分钟为子subKey，总计365个key，每个key有24*60=1440个subKey
                    string key = "STATHTTP_" + TransForm.DateTimeFormat(dt, "yyyyMMdd");
                    string subKey = TransForm.DateTimeFormat(dt, "HHmm");
                    Task<long> task = redisSTAT.HashIncrementAsync(key, subKey, secondCount);

                    //设置过期时间
                    if (!key.Equals(keyPre))
                    {
                        redisSTAT.KeyExpire(key, TimeSpan.FromDays(365 * 2));
                    }
                    keyPre = key;
                }
                catch { }
                finally
                {
                    int gap = (int)sw.ElapsedMilliseconds;
                    Thread.Sleep(gap >= interval ? 0 : (interval - gap));
                }
            }
        }
    }
}
